The runtime serves a list of modules on one credential, naming a bundle that fails to load (hq ADR 0175, to-be 38 WP1)
One tool runtime per node, host-side, is what the runtime was written to be; the catalogue built a container per module around it instead. This lets `serve` take a list — MESH_TOOL_MODULES as <module>=<entrypoint> entries — and do for every assigned module what it did for one: read that module's membership and follow it, serve its tools where the membership says, serve each held seat's verbs on the seat's subjects. The seats come from the memberships now, so the node's credential carries no claims; a module's own runtime still reads its credential's, so nothing built today changes behaviour. A bare path in MESH_TOOL_MODULES stays the one-module form. A bundle that throws on import is said in the log and in what `tools` answers for its module (`failed`), which discovery lists with the reason instead of as "not answering"; the other bundles serve. The filter that dropped every registration under a name but the one module goes; what stays is that a registration under a seat's name is served only where some served module claims the seat. A tool runs attributed to its module, so an event it emits lands on the module's subject and not the runtime's. MESH_OPERATOR_ACCOUNT and MESH_OPERATOR_HOME are read and said; tools take them from their environment. Proven against a real bus: three bundles, one broken; five tools and two seat verbs answer on their subjects; `tools` names the failed bundle; a membership re-issued mid-run re-serves.
This commit is contained in:
jochen
+228
-92
@@ -1,14 +1,20 @@
|
||||
// The tool runtime — the thin per-node process that makes a module's tools actually serve. It
|
||||
// binds the mesh broker, loads the assigned modules' tool entrypoints (each of which calls
|
||||
// registerModuleTools as it imports), and hands them to the sdk's serving harness. Everything hard
|
||||
// — dispatch, collection, duplicate-name safety — is the sdk's; this is the wrapper.
|
||||
// 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 { pathToFileURL } from "node:url";
|
||||
import { resolve } from "node:path";
|
||||
import { useBroker } from "@novox/mesh-sdk/messaging";
|
||||
import { collectTools, toolKey } from "@novox/mesh-sdk/tools";
|
||||
import { collectTools, toolKey, type ToolDefinition } from "@novox/mesh-sdk/tools";
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
import { seatToolSubject, type Credential, type RuntimeBroker } from "./broker-nats.js";
|
||||
import { atWork, seatToolSubject, type Credential, type RuntimeBroker } from "./broker-nats.js";
|
||||
|
||||
/**
|
||||
* The one verb every module's runtime answers for it (novox/hq ADR 0152, design 34 §3): the
|
||||
@@ -28,137 +34,267 @@ export interface ToolsAnswer {
|
||||
* 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;
|
||||
/** Absolute paths to the assigned modules' compiled tool entrypoints (e.g. .../umami/tools/index.js). */
|
||||
moduleEntrypoints: string[];
|
||||
/** 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. */
|
||||
* (novox/hq ADR 0159). Absent for a runtime started by hand, which then serves no seat its
|
||||
* memberships do not name. */
|
||||
credential?: Credential;
|
||||
}
|
||||
|
||||
/** 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);
|
||||
|
||||
for (const entry of opts.moduleEntrypoints) {
|
||||
// Importing the entrypoint runs its registerModuleTools(...) — that is the whole handshake.
|
||||
await import(pathToFileURL(resolve(entry)).href);
|
||||
// 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]);
|
||||
}
|
||||
|
||||
// Serve the RPC endpoint only if a module actually registered a tool. A pure-events module (the
|
||||
// audit logger) registers none, and its scoped account may not declare the serve queue — so a
|
||||
// runtime that always served would fail for exactly the modules that never needed it.
|
||||
// A registration 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, never as a module's
|
||||
// tools and never listed among them. Everything else is the module's own.
|
||||
// A module named like its seat (the catalogue is the mesh-catalog seat) registers once and is
|
||||
// both: its tools are the module's and the seat's verbs alike.
|
||||
const self = opts.credential?.module;
|
||||
const seatNames = new Set((opts.credential?.claims ?? []).map((c) => c.seat));
|
||||
// A registration under a name that is neither this module nor a seat it claims is not served:
|
||||
// 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 ownRegistrations = collectTools().filter(({ module }) => {
|
||||
if (module === self || !self || seatNames.has(module)) return module === self || !self;
|
||||
console.log(`[mesh-tools] ${self} registers tools under "${module}", which is neither this module nor a seat its credential claims; not served until the mesh issues the claim`);
|
||||
return false;
|
||||
});
|
||||
const tools = ownRegistrations.flatMap(({ module, tools: own }) => own.map((t) => ({ module, name: t.name })));
|
||||
const stops: Array<() => void> = [];
|
||||
const stop = (): void => stops.splice(0).forEach((s) => s());
|
||||
// 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 membership (ADR 0160).
|
||||
for (const { module, tools: own } of ownRegistrations) {
|
||||
const seen = new Set<string>();
|
||||
for (const t of own) {
|
||||
if (seen.has(t.name)) {
|
||||
stop();
|
||||
throw new Error(`${module} exposes two tools named ${t.name} — refused`);
|
||||
// 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);
|
||||
}
|
||||
|
||||
// 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.
|
||||
const failed = new Map<string, string>();
|
||||
const owner: string[] = []; // registration index → the module whose bundle registered it
|
||||
for (const [module, entrypoints] of served) {
|
||||
for (const entry of entrypoints) {
|
||||
const before = collectTools().length;
|
||||
try {
|
||||
await import(pathToFileURL(resolve(entry)).href);
|
||||
} 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`);
|
||||
}
|
||||
seen.add(t.name);
|
||||
stops.push(await opts.broker.handle(toolKey(module, t.name), (args: Record<string, unknown> | undefined) => t.run(args ?? {})));
|
||||
const after = collectTools().length;
|
||||
for (let i = before; i < after; i++) owner[i] = module;
|
||||
}
|
||||
}
|
||||
|
||||
// And, for every module that serves any, the verb that says what it serves. 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.
|
||||
const runtime = opts.broker as RuntimeBroker;
|
||||
// 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);
|
||||
const registrations = collectTools().map((r, i) => ({ ...r, owner: owner[i] ?? self ?? r.module }));
|
||||
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 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.length === 0) continue;
|
||||
if (own.some((t) => t.name === TOOLS_VERB)) {
|
||||
stop();
|
||||
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 subjectsOf = (tool: string): string[] | undefined => {
|
||||
const issued = typeof runtime.membership === "function" ? runtime.membership() : undefined;
|
||||
if (!issued) return undefined;
|
||||
const plain = issued.serves.filter((s) => s.queue).map((s) => s.subject.replace("{tool}", tool));
|
||||
const mine = issued.serves.filter((s) => !s.queue).map((s) => s.subject.replace("{tool}", tool));
|
||||
return [...plain, ...mine];
|
||||
};
|
||||
const answer: ToolsAnswer = {
|
||||
module,
|
||||
tools: own.map((t) => ({ name: t.name, description: t.description, input: t.input, subjects: subjectsOf(t.name) })),
|
||||
};
|
||||
stops.push(await opts.broker.handle(toolKey(module, TOOLS_VERB), async () => answer));
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[mesh-tools] serving ${tools.length} tool(s): ${tools.map((t) => t.name).join(", ") || "(none)"}`);
|
||||
stops.push(await serveClaimedSeats(opts.broker as RuntimeBroker, opts.credential));
|
||||
return () => {
|
||||
for (const s of stops) s();
|
||||
};
|
||||
// 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). The credential names the
|
||||
* seats this module claims and the verbs each promises; each verb is served on the seat's own
|
||||
* subject by the module's tool of the same 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.
|
||||
* 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, credential?: Credential): Promise<() => void> {
|
||||
const claims = credential?.claims ?? [];
|
||||
if (claims.length === 0 || typeof broker.handleSubject !== "function") return () => {};
|
||||
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, tools } of collectTools()) {
|
||||
if (!claims.some((c) => c.seat === module)) continue;
|
||||
const verbs = new Map<string, (args: Record<string, unknown>) => Promise<unknown>>();
|
||||
for (const t of tools) verbs.set(t.name, (args) => t.run(args));
|
||||
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 = [];
|
||||
const issued = typeof broker.membership === "function" ? broker.membership() : undefined;
|
||||
for (const claim of claims) {
|
||||
const verbs = implementations.get(claim.seat);
|
||||
for (const verb of claim.serves ?? []) {
|
||||
const run = verbs?.get(verb);
|
||||
if (!run) {
|
||||
console.log(`[mesh-tools] claims ${claim.seat} and implements no ${verb}, which that seat promises; not served`);
|
||||
continue;
|
||||
}
|
||||
// Where the mesh issued the verb when it has; the derived shape until then.
|
||||
const subject = issued?.seats?.find((s) => s.seat === claim.seat && s.verb === verb)?.subject
|
||||
?? seatToolSubject(claim.seat, verb, claim.scope, credential?.node);
|
||||
stops.push(await broker.handleSubject(subject, run));
|
||||
console.log(`[mesh-tools] serving ${claim.seat}'s ${verb} on ${subject}, admitted where this module holds the seat`);
|
||||
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());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user