novox/mesh-tools
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A bundle that is not JavaScript is launched and spoken to over MCP on stdio (hq ADR 0187, to-be 38 WP1b)

Browse Source 
The runtime imported a bundle into its own process, which only JavaScript can be. Now an entrypoint
that is not a plain JavaScript file — or is one marked executable — is started as a child with the
runtime's environment and asked `tools/list` once and `tools/call` per call; what it lists is
registered exactly as an imported bundle's registrations are, a `<seat>.<verb>` name as the seat's
implementation. So a tools bundle may be in any language, and the mesh's part — the subjects, the
seats, the `tools` answer, a failed bundle named — stays in the runtime and is shared by all of
them. A child that exits mid-call tells the caller so and is started again on its next call.

Proven against a real bus beside the three bundles already there: a Python bundle with no SDK at
all answers its tool and its seat verb; a TypeScript bundle written against the protocol and marked
executable is served through the launcher, shortcut off; a bundle told to exit is relaunched.
This commit is contained in:
jochen
2026-10-02 21:24:20 +02:00
parent 6390d1d7fb
commit 1436b02755
6 changed files with 255 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/launch.ts
+169
View File
@@ -0,0 +1,169 @@
// A tools bundle as a process the runtime launches (novox/hq ADR 0187).
//
// The runtime does not run a tool's code itself when the bundle is not JavaScript: it starts the
// bundle's executable as a child with the runtime's environment and speaks MCP over stdio to it —
// `initialize`, `tools/list` once, `tools/call` per call. A Rust binary, a Go binary, a Python
// script and a Node script are the same thing from here: a process that answers those. Everything
// the mesh adds — the subjects from the membership, the held seats, the `tools` answer, a bundle
// that failed named and the others serving — is the runtime's, outside this file.
//
// A tool the child lists as `<seat>.<verb>` is the module's implementation of that seat's verb;
// any other name is the module's own tool. The same rule the in-process registration follows.
import { spawn, type ChildProcess } from "node:child_process";
import { accessSync, constants } from "node:fs";
import type { ToolDefinition } from "@novox/mesh-sdk/tools";
/** The protocol version this speaks; a bundle says the same. */
export const PROTOCOL = "2025-03-26";
/** How long a child has to answer `initialize` and `tools/list` before it is a failed bundle, and
* how long a call may take before the caller is told the tool is slow rather than absent. */
const HANDSHAKE_MS = 10_000;
const CALL_MS = 30_000;
/** Whether an entrypoint is launched as a process rather than imported: anything that is not a
* plain JavaScript file, and a JavaScript file marked executable — a bundle written against the
* protocol in TypeScript, served the same way as any other language. */
export function launches(entry: string): boolean {
const javascript = /\.(m|c)?js$/.test(entry);
let executable = false;
try {
accessSync(entry, constants.X_OK);
executable = true;
} catch {
// not executable, or not there — importing will say which
}
return !javascript || executable;
}
/** What a launched bundle registers: the groups the in-process path would have, by name. */
export interface Launched {
registrations: { module: string; tools: ToolDefinition[] }[];
stop(): void;
}
interface Pending {
resolve(v: any): void;
reject(e: Error): void;
timer: NodeJS.Timeout;
}
/**
* Launch a bundle and learn its tools. Rejects when the child cannot be started or does not complete
* the handshake, which the runtime records as the bundle having failed. A child that exits later is
* started again on the next call, once; a call in flight when it died is told so.
*/
export async function launch(module: string, entry: string, env: NodeJS.ProcessEnv = process.env): Promise<Launched> {
let child: ChildProcess | undefined;
let nextId = 1;
const pending = new Map<number, Pending>();
let stopped = false;
const start = async (): Promise<void> => {
const proc = spawn(entry, [], { stdio: ["pipe", "pipe", "pipe"], env });
child = proc;
let buffered = "";
proc.stdout!.on("data", (chunk: Buffer) => {
buffered += chunk.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) continue;
let reply: { id?: number; result?: unknown; error?: { message?: string } };
try {
reply = JSON.parse(line);
} catch {
console.log(`[mesh-tools] ${module}'s bundle said something that is not a reply: ${line.slice(0, 120)}`);
continue;
}
const waiting = typeof reply.id === "number" ? pending.get(reply.id) : undefined;
if (!waiting) continue;
pending.delete(reply.id!);
clearTimeout(waiting.timer);
if (reply.error) waiting.reject(new Error(reply.error.message ?? "the bundle refused the request"));
else waiting.resolve(reply.result);
}
});
// stderr is the bundle's log; kept under the module's name so a fault reads where it belongs.
proc.stderr!.on("data", (chunk: Buffer) => {
for (const line of chunk.toString("utf8").split("\n")) if (line.trim()) console.log(`[${module}] ${line}`);
});
const exited = new Promise<never>((_, reject) => {
proc.once("error", (err) => reject(err));
proc.once("exit", (code, signal) => {
const why = `${module}'s bundle exited (${signal ?? code})`;
for (const [id, p] of pending) {
pending.delete(id);
clearTimeout(p.timer);
p.reject(new Error(why));
}
if (child === proc) child = undefined;
if (!stopped) console.log(`[mesh-tools] ${why}; started again on its next call`);
reject(new Error(why));
});
});
const ask = (method: string, params: unknown, ms: number): Promise<any> =>
Promise.race([
new Promise<any>((resolve, reject) => {
const id = nextId++;
const timer = setTimeout(() => {
pending.delete(id);
reject(new Error(`${module}'s bundle did not answer ${method} in ${ms / 1000}s`));
}, ms);
pending.set(id, { resolve, reject, timer });
proc.stdin!.write(JSON.stringify({ jsonrpc: "2.0", id, method, params }) + "\n");
}),
exited,
]);
exited.catch(() => {}); // observed through the race; never unhandled
(proc as ChildProcess & { ask?: typeof ask }).ask = ask;
await ask("initialize", { protocolVersion: PROTOCOL, capabilities: {}, clientInfo: { name: "node-tools", version: "1" } }, HANDSHAKE_MS);
proc.stdin!.write(JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }) + "\n");
};
const asking = async (method: string, params: unknown, ms: number): Promise<any> => {
if (!child) await start();
return (child as ChildProcess & { ask: (m: string, p: unknown, ms: number) => Promise<any> }).ask(method, params, ms);
};
await start();
const listed = (await asking("tools/list", {}, HANDSHAKE_MS)) as { tools?: { name: string; description?: string; inputSchema?: unknown }[] };
const groups = new Map<string, ToolDefinition[]>();
for (const t of listed.tools ?? []) {
const dot = t.name.indexOf(".");
const under = dot < 0 ? module : t.name.slice(0, dot);
const name = dot < 0 ? t.name : t.name.slice(dot + 1);
const tools = groups.get(under) ?? [];
tools.push({
name,
description: t.description ?? "",
input: (t.inputSchema as Record<string, unknown> | undefined) ?? {},
run: async (args) => {
const result = (await asking("tools/call", { name: t.name, arguments: args ?? {} }, CALL_MS)) as {
content?: { type: string; text?: string }[];
isError?: boolean;
};
const text = result?.content?.find((c) => c.type === "text")?.text ?? "";
if (result?.isError) throw new Error(text || `${module}.${t.name} failed`);
// The bundle's answer is JSON as text (that is what every MCP host renders); handed back as
// the value it encodes so a caller on the bus sees what an in-process tool would return.
try {
return JSON.parse(text);
} catch {
return text;
}
},
});
groups.set(under, tools);
}
return {
registrations: [...groups].map(([under, tools]) => ({ module: under, tools })),
stop: () => {
stopped = true;
child?.kill("SIGTERM");
child = undefined;
},
};
}
src/runtime.ts
+21 -6
View File
@@ -15,6 +15,7 @@ import { useBroker } from "@novox/mesh-sdk/messaging";
import { collectTools, toolKey, type ToolDefinition } from "@novox/mesh-sdk/tools";
import type { Broker } from "@novox/mesh-sdk/messaging";
import { atWork, seatToolSubject, type Credential, type RuntimeBroker } from "./broker-nats.js";
import { launch, launches } from "./launch.js";
/**
* The one verb every module's runtime answers for it (novox/hq ADR 0152, design 34 §3): the
@@ -106,20 +107,31 @@ export async function runTools(opts: RuntimeOptions): Promise<() => void> {
// Importing the entrypoint runs its registerModuleTools(...) — that is the whole handshake — and
// the registrations it adds are the ones that appear after it, which is how each is attributed
// to the module whose bundle made it.
// A bundle that is not plain JavaScript — or is marked executable — is launched as a process
// and spoken to over MCP on stdio instead (ADR 0187); what it lists is registered the same way.
const failed = new Map<string, string>();
const owner: string[] = []; // registration index → the module whose bundle registered it
const launched: { module: string; owner: string; tools: ToolDefinition[] }[] = [];
const children: Array<() => void> = [];
for (const [module, entrypoints] of served) {
for (const entry of entrypoints) {
const before = collectTools().length;
const path = resolve(entry);
try {
await import(pathToFileURL(resolve(entry)).href);
if (launches(path)) {
const child = await launch(module, path);
children.push(child.stop);
for (const r of child.registrations) launched.push({ ...r, owner: module });
continue;
}
const before = collectTools().length;
await import(pathToFileURL(path).href);
const after = collectTools().length;
for (let i = before; i < after; i++) owner[i] = module;
} catch (err) {
const why = err instanceof Error ? err.message : String(err);
failed.set(module, why);
console.log(`[mesh-tools] ${module}'s bundle ${entry} failed to load: ${why}; its tools are not served here`);
}
const after = collectTools().length;
for (let i = before; i < after; i++) owner[i] = module;
}
}
@@ -131,14 +143,17 @@ export async function runTools(opts: RuntimeOptions): Promise<() => void> {
// out rather than fatal — on 2026-10-01 the credential of a module that had just learned to
// implement a seat did not yet name the claim, and the whole runtime restarted for it.
const claimed = seatsClaimed(served.keys(), self, opts.credential, runtime);
const registrations = collectTools().map((r, i) => ({ ...r, owner: owner[i] ?? self ?? r.module }));
const registrations = [
...collectTools().map((r, i) => ({ ...r, owner: owner[i] ?? self ?? r.module })),
...launched,
];
const ownRegistrations = registrations.filter(({ module, owner: by }) => {
if (served.has(module)) return true;
if (claimed.has(module)) return false;
console.log(`[mesh-tools] ${by} registers tools under "${module}", which is neither a module served here nor a seat one of them claims; not served until the mesh issues the claim`);
return false;
});
const stops: Array<() => void> = [];
const stops: Array<() => void> = [...children];
const stop = (): void => stops.splice(0).forEach((s) => s());
// Refused before anything is bound if a module named a tool of its own `tools`: one name