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

node-tools launches every bundle it serves; a child's emit is published as its module (hq ADR 0193)

Browse Source 
Every served entrypoint is started as a process speaking MCP over stdio, told its module and node;
one that is not executable is refused by name. The import path, the SDK resolve hook (issue 209)
and the per-registration hand-off go. The one-module form the per-module containers use is still
imported until they move (to-be 38 WP4c). A child's mesh/publish is published as its module and
answered once accepted; a child that dies says why in its own last words. Fixtures are served
through launchers exactly as the builder writes them.
This commit is contained in:
jochen
2026-10-03 21:12:15 +02:00
parent 26e7fcb4cb
commit b182943c24
14 changed files with 146 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/src/launch.ts
+27 -4
View File
@@ -13,6 +13,8 @@
import { spawn, type ChildProcess } from "node:child_process";
import { accessSync, constants } from "node:fs";
import type { ToolDefinition } from "@novox/mesh-sdk/tools";
import { broker, type Envelope } from "@novox/mesh-sdk/messaging";
import { atWork } from "./broker-nats.js";
/** The protocol version this speaks; a bundle says the same. */
export const PROTOCOL = "2025-03-26";
@@ -71,29 +73,50 @@ export async function launch(module: string, entry: string, env: NodeJS.ProcessE
const line = buffered.slice(0, at).trim();
buffered = buffered.slice(at + 1);
if (!line) continue;
let reply: { id?: number; result?: unknown; error?: { message?: string } };
let reply: { id?: number | string; method?: string; params?: unknown; 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;
}
// **The bundle asks the runtime to emit** (novox/hq ADR 0193): published on the bus as this
// module, and answered once the bus has accepted it, so the tool's emit means what it means
// in-process. Nothing else a bundle may ask.
if (typeof reply.method === "string") {
const id = reply.id;
const answer = (m: Record<string, unknown>) => proc.stdin!.write(JSON.stringify({ jsonrpc: "2.0", id, ...m }) + "\n");
if (reply.method !== "mesh/publish") {
if (id !== undefined) answer({ error: { code: -32601, message: `the runtime answers no ${reply.method} from a bundle` } });
continue;
}
atWork.run({ module }, () => broker().publish(reply.params as Envelope<unknown>))
.then(() => { if (id !== undefined) answer({ result: {} }); })
.catch((err: unknown) => { if (id !== undefined) answer({ error: { code: -32000, message: err instanceof Error ? err.message : String(err) } }); });
continue;
}
const waiting = typeof reply.id === "number" ? pending.get(reply.id) : undefined;
if (!waiting) continue;
pending.delete(reply.id!);
pending.delete(reply.id as number);
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.
// The last thing it said is kept, so a bundle that dies says why in its own words, not by code.
let lastSaid = "";
proc.stderr!.on("data", (chunk: Buffer) => {
for (const line of chunk.toString("utf8").split("\n")) if (line.trim()) console.log(`[${module}] ${line}`);
for (const line of chunk.toString("utf8").split("\n")) {
if (!line.trim()) continue;
console.log(`[${module}] ${line}`);
if (/\S/.test(line) && !/^\s+at\s/.test(line) && !/^Node\.js v/.test(line)) lastSaid = line.trim();
}
});
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})`;
const why = `${module}'s bundle exited (${signal ?? code})` + (lastSaid ? `: ${lastSaid}` : "");
for (const [id, p] of pending) {
pending.delete(id);
clearTimeout(p.timer);
node-tools/src/runtime.ts
+27 -93
View File
@@ -9,12 +9,9 @@
// and the per-module shape is the list with one entry. A bundle that fails to import is named —
// in the log and in what `tools` answers for it — and the others serve.
import { fileURLToPath, pathToFileURL } from "node:url";
import { dirname, join, resolve } from "node:path";
import { existsSync, readFileSync } from "node:fs";
import { registerHooks } from "node:module";
import { pathToFileURL } from "node:url";
import { resolve } from "node:path";
import { useBroker } from "@novox/mesh-sdk/messaging";
import * as sdkTools from "@novox/mesh-sdk/tools";
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";
@@ -115,6 +112,7 @@ export async function runTools(opts: RuntimeOptions): Promise<() => void> {
for (const s of opts.serves ?? []) {
served.set(s.module, [...(served.get(s.module) ?? []), ...s.entrypoints]);
}
const ownEntrypoints = opts.moduleEntrypoints ?? [];
if (opts.moduleEntrypoints?.length) {
if (!self) {
throw new Error(
@@ -143,34 +141,38 @@ export async function runTools(opts: RuntimeOptions): Promise<() => void> {
const envs = opts.envs ?? new Map<string, Record<string, string>>();
const envFor = (module: string): NodeJS.ProcessEnv => ({ ...process.env, ...(envs.get(module) ?? {}) });
// Import each bundle, guarded (ADR 0175: one faulty bundle must not take the node's tools down).
// 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 0188); what it lists is registered the same way.
// Every bundle's import of the SDK resolves to this runtime's copy (04-ISSUES/209): one registry
// of tools, one broker. Installed before the first bundle is imported.
oneSdk();
// Every bundle this runtime serves is launched as a process and spoken to over MCP on stdio (ADR
// 0188, ADR 0193): given the runtime's words and its module's own, and told the module it serves
// it as. The runtime knows no language; an entrypoint that is not executable was not built to be
// served, and is refused by name. A bundle that fails to start is named, and the others serve
// (ADR 0175: one faulty bundle must not take the node's tools down).
//
// The one-module form — the credential's own module's entrypoints, which the per-module containers
// still use for their event handlers and provisioners until they move (to-be 38 WP4c) — is imported
// into this process as before: one module, one SDK, its container's own environment.
// The machine this runtime serves, which an event a launched tool emits is stamped with.
const node = opts.credential?.node ?? (typeof (runtime as unknown as { node?: unknown }).node === "string" ? (runtime as unknown as { node: string }).node : undefined);
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) {
const imported = module === self && ownEntrypoints.length > 0;
for (const entry of entrypoints) {
const path = resolve(entry);
try {
if (launches(path)) {
// Told the module it serves it as, so a seat's verbs are the seat's (ADR 0193).
const child = await launch(module, path, { ...envFor(module), MESH_SERVED_MODULE: module });
children.push(child.stop);
for (const r of child.registrations) launched.push({ ...r, owner: module });
if (imported) {
await import(pathToFileURL(path).href);
continue;
}
const before = collectTools().length;
await import(pathToFileURL(path).href);
const after = collectTools().length;
for (let i = before; i < after; i++) owner[i] = module;
if (!launches(path)) {
throw new Error(`${path} is not executable; a bundle the runtime serves is started, never imported, and its build makes it executable (novox/hq ADR 0193)`);
}
const child = await launch(module, path, {
...envFor(module), MESH_SERVED_MODULE: module, MESH_MODULE: module,
...(node ? { MESH_NODE: node } : {}),
});
children.push(child.stop);
for (const r of child.registrations) launched.push({ ...r, owner: module });
} catch (err) {
const why = err instanceof Error ? err.message : String(err);
failed.set(module, why);
@@ -187,16 +189,8 @@ 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);
// Each registration's contributor is given its own module's environment and no other's (ADR 0192):
// the runtime's own words, and over them what the mesh composed for the module whose bundle made
// the registration. An SDK too old to ask per registration cannot do that; said, not hidden.
const each = (sdkTools as { collectToolsEach?: (f: (module: string, i: number) => NodeJS.ProcessEnv) => { module: string; tools: ToolDefinition[] }[] }).collectToolsEach;
if (!each && envs.size > 0) {
console.log(`[mesh-tools] this runtime's SDK cannot give each bundle its own environment; ${[...envs.keys()].join(", ")} serve with the runtime's words only (novox/hq ADR 0192)`);
}
const collected = each ? each((module, i) => envFor(owner[i] ?? self ?? module)) : collectTools();
const registrations = [
...collected.map((r, i) => ({ ...r, owner: owner[i] ?? self ?? r.module })),
...collectTools().map((r) => ({ ...r, owner: self ?? r.module })),
...launched,
];
const ownRegistrations = registrations.filter(({ module, owner: by }) => {
@@ -365,63 +359,3 @@ async function serveClaimedSeats(
if (typeof broker.onMembership === "function") broker.onMembership(() => void serve());
return () => stops.forEach((s) => s());
}
let sdkHooked = false;
const SDK = "@novox/mesh-sdk";
/**
* The one SDK in a node's runtime (novox/hq ADR 0175, 04-ISSUES/209).
*
* A bundle carries its own dependencies — the toolchain copies them in so a bundle starts anywhere
* (to-be 38 WP3) — and among them is a copy of the SDK. Imported in this process, that copy would be
* a second SDK: its own registry of tools, its own broker handle. A bundle calling registerModuleTools
* through it registers into a list this runtime never reads, and its tools are silently not served.
* So every import of the SDK, from whichever bundle, is resolved as if this runtime had written it:
* one registry, one broker — the runtime's. Everything else a bundle carries resolves from the
* bundle's own tree, as before. A launched bundle (ADR 0188) is another process and is untouched.
*
* Installed once, in-thread, before the first bundle is imported; the hook sees every import after,
* `require` included. A bundle whose own copy is another version than the runtime's is said once,
* so a tool failing against the runtime's SDK points at the bundle rather than at the runtime.
*/
function oneSdk(): void {
if (sdkHooked) return;
registerHooks({
resolve(specifier, context, next) {
if (specifier === SDK || specifier.startsWith(SDK + "/")) {
if (context.parentURL) sayOtherSdk(context.parentURL);
return next(specifier, { ...context, parentURL: import.meta.url });
}
return next(specifier, context);
},
});
sdkHooked = true;
}
const sdkSaid = new Set<string>();
/** The version of the SDK copy nearest a file, by its package.json, or nothing when the file has none above it. */
function sdkVersionNear(fileURL: string): { dir: string; version: string } | undefined {
let dir = dirname(fileURLToPath(fileURL));
for (;;) {
const pkg = join(dir, "node_modules", SDK, "package.json");
if (existsSync(pkg)) {
try {
return { dir, version: String((JSON.parse(readFileSync(pkg, "utf8")) as { version?: string }).version ?? "?") };
} catch {
return { dir, version: "?" };
}
}
const up = dirname(dir);
if (up === dir) return undefined;
dir = up;
}
}
function sayOtherSdk(parentURL: string): void {
if (!parentURL.startsWith("file:")) return;
const own = sdkVersionNear(import.meta.url);
const theirs = sdkVersionNear(parentURL);
if (!theirs || !own || theirs.dir === own.dir || sdkSaid.has(theirs.dir)) return;
sdkSaid.add(theirs.dir);
if (theirs.version !== own.version) {
console.log(`[mesh-tools] ${theirs.dir} carries ${SDK} ${theirs.version}; this runtime's is ${own.version}, and the bundle speaks to the runtime's`);
}
}