jochen
193ed0ac63
The mesh composes every served module's words into MESH_TOOL_ENV; the runtime takes it at start and removes it from its own environment, then gives each registration's contributor and each launched child the runtime's words plus its own module's, never another's. Against an SDK without collectToolsEach it says so and serves with the runtime's words only. The test serves two imported bundles and one launched, each answering with its own words and none of the others'.
427 lines
21 KiB
TypeScript
427 lines
21 KiB
TypeScript
// The tool runtime — the per-node process that makes the mesh's tools actually serve (novox/hq
|
|
// ADR 0175). It binds the mesh broker, loads the served modules' tool bundles (each of which calls
|
|
// registerModuleTools as it imports), and serves every module's tools on that module's subjects and
|
|
// every held seat's verbs on the seat's. Everything hard — dispatch, collection, duplicate-name
|
|
// safety — is the sdk's; this is the wrapper.
|
|
//
|
|
// One runtime, many modules. It was written for one module per process and ran that way in a
|
|
// container per module; it now serves a list, as the one process per node the host supervises,
|
|
// 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 { 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";
|
|
import { launch, launches } from "./launch.js";
|
|
|
|
/**
|
|
* The one verb every module's runtime answers for it (novox/hq ADR 0152, design 34 §3): the
|
|
* module's tool names, descriptions and argument schemas, from the code that answers them and
|
|
* from nowhere else. Discovery asks the module, because a copy kept anywhere else drifts.
|
|
*/
|
|
export const TOOLS_VERB = "tools";
|
|
|
|
/** What `tools` answers for one module. */
|
|
export interface ToolsAnswer {
|
|
module: string;
|
|
tools: {
|
|
name: string;
|
|
description: string;
|
|
input: Readonly<Record<string, unknown>>;
|
|
/** Where this tool is answered, as the mesh issued it (ADR 0160): the module's plain subject
|
|
* first when there is one, then this machine's. A caller composes nothing. */
|
|
subjects?: string[];
|
|
}[];
|
|
/** Why this module serves nothing here, when its bundle failed to load (ADR 0175): said where
|
|
* discovery looks, so a module that is silent and one that is broken are told apart. */
|
|
failed?: string;
|
|
}
|
|
|
|
/** One module this runtime serves: its name and its compiled tool entrypoints. */
|
|
export interface ServedModule {
|
|
module: string;
|
|
/** Absolute paths to the module's compiled tool entrypoints (e.g. .../umami/tools/index.js). */
|
|
entrypoints: string[];
|
|
}
|
|
|
|
export interface RuntimeOptions {
|
|
/** The mesh broker to serve over. */
|
|
broker: Broker;
|
|
/** The modules to serve, each with its entrypoints. */
|
|
serves?: ServedModule[];
|
|
/** The credential's own module's entrypoints — the one-module form, which the per-module
|
|
* containers still use; the same as naming the credential's module in `serves`. */
|
|
moduleEntrypoints?: string[];
|
|
/** The credential the mesh delivered, for what it says about the seats this module claims
|
|
* (novox/hq ADR 0159). Absent for a runtime started by hand, which then serves no seat its
|
|
* memberships do not name. */
|
|
credential?: Credential;
|
|
/** What each served module's bundles are given (novox/hq ADR 0192): module → words, composed by
|
|
* the mesh per machine. A module absent here is given the runtime's own words and nothing more. */
|
|
envs?: ReadonlyMap<string, Readonly<Record<string, string>>>;
|
|
}
|
|
|
|
/** The variable the mesh composes every served module's environment into, as JSON (ADR 0192). Read
|
|
* once at start and removed from the process's environment, so no bundle finds another's there. */
|
|
export const TOOL_ENV = "MESH_TOOL_ENV";
|
|
|
|
/** Read and remove the composed environments from an environment (the process's, by default). */
|
|
export function takeToolEnvs(env: NodeJS.ProcessEnv = process.env): Map<string, Record<string, string>> {
|
|
const raw = env[TOOL_ENV];
|
|
delete env[TOOL_ENV];
|
|
const out = new Map<string, Record<string, string>>();
|
|
if (!raw) return out;
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(raw);
|
|
} catch {
|
|
throw new Error(`${TOOL_ENV} is not JSON; the mesh composes it as {"<module>": {"<word>": "<value>"}}`);
|
|
}
|
|
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
|
|
throw new Error(`${TOOL_ENV} is not an object of modules`);
|
|
}
|
|
for (const [module, words] of Object.entries(parsed as Record<string, unknown>)) {
|
|
if (!words || typeof words !== "object" || Array.isArray(words)) {
|
|
throw new Error(`${TOOL_ENV}: ${module}'s environment is not an object of words`);
|
|
}
|
|
const own: Record<string, string> = {};
|
|
for (const [k, v] of Object.entries(words as Record<string, unknown>)) own[k] = String(v);
|
|
out.set(module, own);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/** Two environment words the mesh sets for the node's runtime and every tool reads from its
|
|
* environment: whose machine this is (novox/hq to-be 37 §3, ADR 0175). */
|
|
export const OPERATOR_ACCOUNT = "MESH_OPERATOR_ACCOUNT";
|
|
export const OPERATOR_HOME = "MESH_OPERATOR_HOME";
|
|
|
|
/** Load the modules, bind the broker, and serve. Returns a stop function that unhooks serving. */
|
|
export async function runTools(opts: RuntimeOptions): Promise<() => void> {
|
|
useBroker(() => opts.broker);
|
|
const runtime = opts.broker as RuntimeBroker;
|
|
// Whose runtime this is: the credential's module, or the connection's own when a runtime is
|
|
// started by hand without one — the broker was told its module when it connected.
|
|
const self = opts.credential?.module ?? (typeof runtime.module === "string" ? runtime.module : undefined);
|
|
|
|
// What to serve: the list, with the one-module form folded in as the credential's own entry.
|
|
const served = new Map<string, string[]>();
|
|
for (const s of opts.serves ?? []) {
|
|
served.set(s.module, [...(served.get(s.module) ?? []), ...s.entrypoints]);
|
|
}
|
|
if (opts.moduleEntrypoints?.length) {
|
|
if (!self) {
|
|
throw new Error(
|
|
"entrypoints were given with no module to serve them as: name the module (MESH_TOOL_MODULES " +
|
|
"as <module>=<entrypoint>) or connect on a credential that names one",
|
|
);
|
|
}
|
|
served.set(self, [...(served.get(self) ?? []), ...opts.moduleEntrypoints]);
|
|
}
|
|
|
|
// The operator's machine, said once so a tool's behaviour under it can be read back from the
|
|
// log. Tools read the two words from their own environment, which is this process's.
|
|
const account = process.env[OPERATOR_ACCOUNT];
|
|
if (account) {
|
|
console.log(`[mesh-tools] the operator's account here is ${account}` +
|
|
(process.env[OPERATOR_HOME] ? ` (home ${process.env[OPERATOR_HOME]})` : ""));
|
|
}
|
|
|
|
// Follow every served module's membership before loading anything, so what each is issued is
|
|
// known when its tools are bound. A module's own runtime already follows its own.
|
|
if (typeof runtime.follow === "function") {
|
|
for (const module of served.keys()) await runtime.follow(module);
|
|
}
|
|
|
|
// What each module's bundles are given: the runtime's own words, and over them the module's own.
|
|
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();
|
|
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 path = resolve(entry);
|
|
try {
|
|
if (launches(path)) {
|
|
const child = await launch(module, path, envFor(module));
|
|
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`);
|
|
}
|
|
}
|
|
}
|
|
|
|
// A registration under a served module's name is that module's tools, served on its subjects.
|
|
// One under a seat's name is the module's implementation of that seat's verbs (ADR 0159, 0160):
|
|
// served on the seat's subjects by serveClaimedSeats where some served module claims the seat,
|
|
// never as a module's tools and never listed among them. A module named like its seat (the
|
|
// catalogue is the mesh-catalog seat) registers once and is both. Anything else is said and left
|
|
// 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 })),
|
|
...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> = [...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
|
|
// answering two things is the fault nobody can diagnose afterwards, and the runtime is the only
|
|
// place that sees both. Likewise two tools of one module under one name.
|
|
for (const { module, tools: own } of ownRegistrations) {
|
|
if (own.some((t) => t.name === TOOLS_VERB)) {
|
|
throw new Error(
|
|
`${module} names a tool "${TOOLS_VERB}", which is the verb the runtime answers for every ` +
|
|
"module with what it serves (novox/hq ADR 0152) — refused, rename it",
|
|
);
|
|
}
|
|
const seen = new Set<string>();
|
|
for (const t of own) {
|
|
if (seen.has(t.name)) throw new Error(`${module} exposes two tools named ${t.name} — refused`);
|
|
seen.add(t.name);
|
|
}
|
|
}
|
|
|
|
// Each tool on its own key, namespaced by its module (ADR 0047); where that key is answered is
|
|
// the broker's to know from the module's membership (ADR 0160). A tool runs attributed to its
|
|
// module, so what it emits lands on the module's subject and not the runtime's.
|
|
const names: string[] = [];
|
|
for (const { module, tools: own } of ownRegistrations) {
|
|
for (const t of own) {
|
|
names.push(toolKey(module, t.name));
|
|
stops.push(await opts.broker.handle(toolKey(module, t.name), (args: Record<string, unknown> | undefined) =>
|
|
atWork.run({ module }, () => t.run(args ?? {}))));
|
|
}
|
|
}
|
|
|
|
// And, for every served module, the verb that says what it serves — nothing, and why, for a
|
|
// module whose bundle failed. A module that registered nothing and did not fail is a pure-events
|
|
// module (the audit logger), whose scoped account may not declare the serve queue; it is left
|
|
// silent as it always was.
|
|
const byModule = new Map<string, ToolDefinition[]>();
|
|
for (const { module, tools: own } of ownRegistrations) {
|
|
byModule.set(module, [...(byModule.get(module) ?? []), ...own]);
|
|
}
|
|
for (const module of served.keys()) {
|
|
const own = byModule.get(module) ?? [];
|
|
const why = failed.get(module);
|
|
if (own.length === 0 && !why) continue;
|
|
const subjectsOf = (tool: string): string[] | undefined => {
|
|
const m = typeof runtime.membership === "function" ? runtime.membership(module) : undefined;
|
|
if (!m) return undefined;
|
|
const plain = m.serves.filter((s) => s.queue).map((s) => s.subject.replace("{tool}", tool));
|
|
const mine = m.serves.filter((s) => !s.queue).map((s) => s.subject.replace("{tool}", tool));
|
|
return [...plain, ...mine];
|
|
};
|
|
stops.push(await opts.broker.handle(toolKey(module, TOOLS_VERB), async (): Promise<ToolsAnswer> => ({
|
|
module,
|
|
tools: own.map((t) => ({ name: t.name, description: t.description, input: t.input, subjects: subjectsOf(t.name) })),
|
|
...(why ? { failed: why } : {}),
|
|
})));
|
|
}
|
|
|
|
console.log(`[mesh-tools] serving ${names.length} tool(s) for ${served.size} module(s): ${names.join(", ") || "(none)"}` +
|
|
(failed.size ? `; not serving ${[...failed.keys()].join(", ")}, whose bundle(s) failed to load` : ""));
|
|
stops.push(await serveClaimedSeats(runtime, [...served.keys()], self, opts.credential, registrations));
|
|
return () => stop();
|
|
}
|
|
|
|
/** The seats some served module claims: from the credential for its own module, and from every
|
|
* served module's membership (ADR 0160) — the node's runtime holds no claims of its own. */
|
|
function seatsClaimed(
|
|
modules: Iterable<string>,
|
|
self: string | undefined,
|
|
credential: Credential | undefined,
|
|
runtime: RuntimeBroker,
|
|
): Set<string> {
|
|
const out = new Set<string>();
|
|
for (const c of credential?.claims ?? []) if (credential?.module === self) out.add(c.seat);
|
|
for (const module of modules) {
|
|
const m = typeof runtime.membership === "function" ? runtime.membership(module) : undefined;
|
|
for (const s of m?.seats ?? []) out.add(s.seat);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/** One seat's verb, where its callers ask, and which served module holds the seat. */
|
|
interface SeatVerb {
|
|
seat: string;
|
|
verb: string;
|
|
subject: string;
|
|
holder: string;
|
|
}
|
|
|
|
/**
|
|
* Holding a seat means serving its tools (design 33 §3, novox/hq ADR 0159). What a served module
|
|
* claims and promises comes from its membership (ADR 0160) — and, for a module's own runtime, from
|
|
* its credential, which named the claims before memberships did. Each verb is served on the seat's
|
|
* own subject by the tool of the same name registered under the seat's name. Whether this instance
|
|
* *holds* the seat is the bus's to decide: only the holder's account may subscribe the seat's
|
|
* subjects, so a claimant that does not hold it here is refused the subscription and serves nothing
|
|
* — never a failure of its own tools.
|
|
*/
|
|
async function serveClaimedSeats(
|
|
broker: RuntimeBroker,
|
|
served: string[],
|
|
self: string | undefined,
|
|
credential: Credential | undefined,
|
|
registrations: { module: string; owner: string; tools: ToolDefinition[] }[],
|
|
): Promise<() => void> {
|
|
if (typeof broker.handleSubject !== "function") return () => {};
|
|
// A seat's verbs are the role's, not the software's (ADR 0159): implemented under the seat's
|
|
// name — `registerModuleTools("mesh-store", …)` — and never confused with the module's own tools.
|
|
const implementations = new Map<string, Map<string, (args: Record<string, unknown>) => Promise<unknown>>>();
|
|
for (const { module, owner, tools } of registrations) {
|
|
const verbs = implementations.get(module) ?? new Map<string, (args: Record<string, unknown>) => Promise<unknown>>();
|
|
for (const t of tools) verbs.set(t.name, (args) => atWork.run({ module: owner }, () => t.run(args)));
|
|
implementations.set(module, verbs);
|
|
}
|
|
|
|
/** Every verb of every seat a served module claims, where the mesh issued it. */
|
|
const wanted = (): SeatVerb[] => {
|
|
const out: SeatVerb[] = [];
|
|
const have = new Set<string>();
|
|
const add = (v: SeatVerb): void => {
|
|
if (have.has(v.subject)) return;
|
|
have.add(v.subject);
|
|
out.push(v);
|
|
};
|
|
for (const module of served) {
|
|
const m = typeof broker.membership === "function" ? broker.membership(module) : undefined;
|
|
for (const s of m?.seats ?? []) add({ seat: s.seat, verb: s.verb, subject: s.subject, holder: module });
|
|
// The credential's claims, for the module's own runtime: where the mesh issued the verb when
|
|
// it has; the derived shape until then.
|
|
if (module !== self) continue;
|
|
for (const claim of credential?.claims ?? []) {
|
|
for (const verb of claim.serves ?? []) {
|
|
const subject = m?.seats?.find((s) => s.seat === claim.seat && s.verb === verb)?.subject
|
|
?? seatToolSubject(claim.seat, verb, claim.scope, credential?.node);
|
|
add({ seat: claim.seat, verb, subject, holder: module });
|
|
}
|
|
}
|
|
}
|
|
return out;
|
|
};
|
|
|
|
let stops: (() => void)[] = [];
|
|
const serve = async (): Promise<void> => {
|
|
stops.forEach((s) => s());
|
|
stops = [];
|
|
for (const v of wanted()) {
|
|
const run = implementations.get(v.seat)?.get(v.verb);
|
|
if (!run) {
|
|
console.log(`[mesh-tools] ${v.holder} claims ${v.seat} and implements no ${v.verb}, which that seat promises; not served`);
|
|
continue;
|
|
}
|
|
stops.push(await broker.handleSubject(v.subject, run));
|
|
console.log(`[mesh-tools] serving ${v.seat}'s ${v.verb} on ${v.subject}, admitted where ${v.holder} holds the seat`);
|
|
}
|
|
};
|
|
await serve();
|
|
// A membership issued to any served module may add, move or withdraw a seat's verbs.
|
|
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`);
|
|
}
|
|
}
|