316 lines
16 KiB
TypeScript
316 lines
16 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 { pathToFileURL } from "node:url";
|
|
import { resolve } from "node:path";
|
|
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
|
|
* 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;
|
|
}
|
|
|
|
/** 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);
|
|
}
|
|
|
|
// 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.
|
|
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);
|
|
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);
|
|
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> = [...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());
|
|
}
|