node-tools is a module beside mesh-tools: the runtime as a bundle, and serve is the console (hq ADR 0175, to-be 38 WP3)
One repository, two modules (ADR 0069). `node-tools/` holds the runtime — its code, tests, package and the manifest of the module the controller composes a process for on every machine it is assigned to: a bundle of `src/main.js`, the interpreter as a package, a place for the node's credential, the loopback port the console declared, and leave to call every tool. Nothing about how it runs: which bundles to load, where the credential is and whose machine it is are the controller's to compose (WP2). The root module `mesh-tools` keeps the two images TypeScript bundles are compiled in and a module's own service may run in; it is no longer how tools reach a node. As node-tools, `serve` is also the console (ADR 0175 §6): the same process answers MCP on loopback for whoever is on the machine, through which the tools it serves can be called. A module's own runtime in a container keeps serving without a listener. The toolchain image now carries /app/runtime — a package.json saying the compiled files are ES modules and the production node_modules — for the builder to copy into every TypeScript bundle, so a bundle unpacked on a machine starts (ADR 0188 §5; the builder's side is the controller's). Proven here by compiling node-tools with the toolchain's exact flags and starting the result. The AMQP probe script is gone with the bus it probed.
This commit is contained in:
jochen
@@ -0,0 +1,638 @@
|
||||
// The tool runtime's broker client, on NATS.
|
||||
//
|
||||
// **The sdk's contract does not change** (novox/hq ADR 0106, ADR 0039): a module is written
|
||||
// against `request`, `handle`, `publish`, `subscribe`, `close`, and the runtime implements them.
|
||||
// That is why a module built before any of this runs on the new runtime without a rebuild, and
|
||||
// why the sdk's own diff for the whole bus change is three comments.
|
||||
//
|
||||
// Underneath, everything is a subject and durability is JetStream (novox/hq design 25,
|
||||
// design 29).
|
||||
//
|
||||
// mesh.mod.<module>.event.<type> an event this module emits
|
||||
// mesh.mod.<module>.tool.<tool> a tool this module serves
|
||||
// mesh.seat.<seat>.accept.<verb> work submitted to a role
|
||||
//
|
||||
// The module never writes one of those: it names its events and tools locally and the mesh
|
||||
// derives the subject (design 29 §1), so reorganising the subject space leaves every module
|
||||
// correct.
|
||||
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { createHash } from "node:crypto";
|
||||
import net from "node:net";
|
||||
import tls from "node:tls";
|
||||
import { connect as natsConnect, headers as natsHeaders, StringCodec, type JsMsg, type Subscription, type TlsOptions } from "nats";
|
||||
import type { Broker, Envelope, EventHeaders } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
const sc = StringCodec();
|
||||
|
||||
/** Requests wait this long for an answer before failing. Unchanged from what modules already
|
||||
* expect, so a module's timeout handling is not something the bus quietly redefines. */
|
||||
const REQUEST_TIMEOUT_MS = 30_000;
|
||||
|
||||
export class PinMismatchError extends Error {}
|
||||
|
||||
/** A broker credential as the mesh delivers it (novox/hq ADR 0120): the bus's address, the
|
||||
* fingerprint of the certificate it must present, and the node and module the account is scoped
|
||||
* to — the runtime derives its subjects from those rather than being told them. */
|
||||
export interface Credential {
|
||||
url: string;
|
||||
fingerprint?: string;
|
||||
node?: string;
|
||||
module?: string;
|
||||
user?: string;
|
||||
password?: string;
|
||||
/** The seats this module claims, with the verbs each promises (novox/hq ADR 0159). The runtime
|
||||
* serves each claimed seat's verbs with its tools of the same name; the bus admits only the
|
||||
* holder's subscription, so claiming and not holding costs a refused subscription and nothing else. */
|
||||
claims?: { seat: string; scope?: string; serves?: string[] }[];
|
||||
}
|
||||
|
||||
/**
|
||||
* What the mesh issued this assignment (novox/hq ADR 0160): where its tools are served, in which
|
||||
* queue, the verbs of the seats it holds, where its events land, what it may reach. Read from the
|
||||
* ASSIGNMENTS stream at `mesh.assignment.<node>.<module>` — the one subject a runtime derives for
|
||||
* itself — and followed live. Absent for a mesh older than the membership, and then the runtime
|
||||
* serves the shape it always derived, and says so.
|
||||
*/
|
||||
export interface Membership {
|
||||
node: string;
|
||||
module: string;
|
||||
/** Addresses a tool is answered on; `{tool}` stands for the tool's name. */
|
||||
serves: { subject: string; queue?: string }[];
|
||||
seats?: { seat: string; verb: string; subject: string }[];
|
||||
emits: string;
|
||||
reaches?: Record<string, string[]>;
|
||||
tools: string;
|
||||
}
|
||||
|
||||
/** What a tool call answers: the module's own result, and which machine answered it
|
||||
* (novox/hq ADR 0159) — a module on several machines is otherwise an answer from nowhere. */
|
||||
export interface Answered<Res> {
|
||||
result: Res;
|
||||
node?: string;
|
||||
}
|
||||
|
||||
/** The bus as the runtime sees it: the sdk's contract, and the few things only the runtime needs —
|
||||
* an answer that says which machine gave it, serving a subject that is not a module's own tool
|
||||
* (a seat's verb), and following the memberships of the modules it serves beside its own. */
|
||||
export interface RuntimeBroker extends Broker {
|
||||
/** Call a tool by key, or — when `on` names a subject the mesh listed for it (ADR 0160) — there. */
|
||||
ask<Req, Res>(key: string, body: Req, on?: string): Promise<Answered<Res>>;
|
||||
handleSubject<Req, Res>(subject: string, handler: (body: Req) => Promise<Res>): Promise<() => void>;
|
||||
/**
|
||||
* Serve another module's tools from this connection: read its membership on this machine and
|
||||
* follow it, so `handle("<module>.<tool>")` is served where the mesh issued that module (novox/hq
|
||||
* ADR 0175: one runtime per node, every assigned module's tools). The account must be allowed
|
||||
* to read that membership and to subscribe its subjects — the node's is; a module's own is not,
|
||||
* and is refused by the bus, not here.
|
||||
*/
|
||||
follow(module: string): Promise<void>;
|
||||
/** What the mesh issued this assignment — this module's when none is named — or undefined when
|
||||
* nothing has been issued yet. */
|
||||
membership(module?: string): Membership | undefined;
|
||||
/** Called when the mesh issues a new membership to any module this connection follows; the
|
||||
* runtime re-serves on it. The membership says which module it is for. */
|
||||
onMembership(handler: (m: Membership) => void): void;
|
||||
/** The modules whose memberships this connection follows: its own and every one `follow` added. */
|
||||
serving(): string[];
|
||||
/** The module this connection is: what its credential named, and what a bare key serves as. */
|
||||
readonly module: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Which module's tool is at work on this connection, when one is (novox/hq ADR 0175). One runtime
|
||||
* carries many modules' tools, and an event a tool emits must land on the emitting *module's*
|
||||
* subject, not the runtime's — so the runtime runs each tool inside this store, and `publish`
|
||||
* reads the module from it. Outside a tool the connection's own module stands.
|
||||
*/
|
||||
export const atWork = new AsyncLocalStorage<{ module: string }>();
|
||||
|
||||
const ASSIGNMENTS_STREAM = "ASSIGNMENTS";
|
||||
|
||||
/** The one address a runtime derives for itself (ADR 0160). */
|
||||
export function membershipSubject(node: string, module: string): string {
|
||||
return `mesh.assignment.${node}.${module}`;
|
||||
}
|
||||
|
||||
/** Whether a connection failure is worth retrying, or is a fact about this configuration that
|
||||
* retrying cannot change. The runtime's supervisor asks this and does not need to know what it
|
||||
* is connected to. */
|
||||
export function fatalBrokerReason(err: unknown): string | null {
|
||||
if (err instanceof PinMismatchError) return "the bus's certificate does not match the pin";
|
||||
const e = err as { code?: string; message?: string };
|
||||
const message = typeof e?.message === "string" ? e.message : String(err);
|
||||
if (e?.code === "ERR_INVALID_URL" || /invalid url/i.test(message)) {
|
||||
return "the bus address is not a usable URL";
|
||||
}
|
||||
if (/authorization violation|user authentication expired|permissions violation/i.test(message)) {
|
||||
return "the bus refused this account";
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to the mesh bus and return a Broker.
|
||||
*
|
||||
* **A module's subjects come from its credential, not from its calls.** `node` and `module` name
|
||||
* the account the mesh issued, and every subject this client publishes or subscribes is derived
|
||||
* from them — so a module cannot name another's namespace even by mistake, and what it emits
|
||||
* matches what the mesh authorised (ADR 0074's identity rule).
|
||||
*/
|
||||
export async function connectNats(
|
||||
target: string | Credential,
|
||||
opts: { module?: string } = {},
|
||||
): Promise<RuntimeBroker> {
|
||||
const cred: Credential = typeof target === "string" ? { url: target } : target;
|
||||
const self = cred.module ?? opts.module;
|
||||
if (!self) {
|
||||
throw new Error(
|
||||
"a broker credential with no module: the runtime derives its subjects from the account " +
|
||||
"the mesh issued, and cannot guess which module it is",
|
||||
);
|
||||
}
|
||||
|
||||
const conn = await natsConnect({
|
||||
servers: cred.url,
|
||||
user: cred.user,
|
||||
pass: cred.password,
|
||||
name: `${cred.node ?? "?"}.${self}`,
|
||||
tls: cred.fingerprint ? await pinnedTls(cred.url, cred.fingerprint) : undefined,
|
||||
// Its own inbox, not a random one: every user's inbox is private to it (design 25 §4), and the
|
||||
// grant names `_INBOX.<user>.>` — a reply space the client invented would be refused, and with
|
||||
// it every pull for the next message and every answer to a tool call.
|
||||
inboxPrefix: cred.user ? `_INBOX.${cred.user}` : undefined,
|
||||
// Reconnect forever: the bus being restarted is an upgrade, not a reason for every module on
|
||||
// the mesh to exit. `close()` stays the only thing that ends the connection.
|
||||
maxReconnectAttempts: -1,
|
||||
});
|
||||
const js = conn.jetstream();
|
||||
|
||||
const subs: Subscription[] = [];
|
||||
// Every registration, and the one reader that dispatches to them. A module has one durable
|
||||
// consumer; the loop belongs to the connection rather than to a subscription.
|
||||
const listeners: { pattern: string; handler: (env: Envelope<unknown>) => Promise<void> }[] = [];
|
||||
let reading: Awaited<ReturnType<Awaited<ReturnType<typeof js.consumers.get>>["consume"]>> | undefined;
|
||||
let closed = false;
|
||||
|
||||
const node = cred.node;
|
||||
|
||||
// The memberships, one per module this connection serves, each read once and followed. A
|
||||
// module's own runtime follows one, its own; the node's runtime follows one per assigned module
|
||||
// (novox/hq ADR 0175) — the same subject shape, the same stream, read as many times as there are
|
||||
// modules, and nothing on the bus learns a new shape for it. A direct get is one request on the
|
||||
// stream's API, which is the whole of what an account may ask JetStream for a subject it is
|
||||
// granted; a 404 is a mesh that has not issued one, which is a fact to say and not an error to
|
||||
// retry.
|
||||
const issued = new Map<string, Membership | undefined>();
|
||||
const issuedHandlers: ((m: Membership) => void)[] = [];
|
||||
const follow = async (module: string): Promise<void> => {
|
||||
if (!node || issued.has(module)) return;
|
||||
issued.set(module, undefined);
|
||||
const subjectOfTheirs = membershipSubject(node, module);
|
||||
try {
|
||||
// The subject-addressed form of a direct get — the stream, then the subject, nothing in
|
||||
// the body — because that is the one address the mesh grants this account on the
|
||||
// stream's API; the body form asks the stream's root, which it may not.
|
||||
const got = await conn.request(`$JS.API.DIRECT.GET.${ASSIGNMENTS_STREAM}.${subjectOfTheirs}`,
|
||||
new Uint8Array(0), { timeout: 5_000 });
|
||||
const status = got.headers?.code ?? 0;
|
||||
if (status === 0 && got.data.length > 0) {
|
||||
issued.set(module, JSON.parse(sc.decode(got.data)) as Membership);
|
||||
}
|
||||
} catch {
|
||||
// Not readable here: an older mesh, a stream not yet asserted, or no grant. Said below.
|
||||
}
|
||||
if (!issued.get(module)) {
|
||||
console.log(`[mesh-tools] no membership issued for ${module} on ${node} yet; serving the derived shape until one arrives`);
|
||||
}
|
||||
try {
|
||||
const live = conn.subscribe(subjectOfTheirs);
|
||||
subs.push(live);
|
||||
void (async () => {
|
||||
for await (const msg of live) {
|
||||
try {
|
||||
const m = JSON.parse(sc.decode(msg.data)) as Membership;
|
||||
issued.set(module, m);
|
||||
console.log(`[mesh-tools] ${module} on ${node} was issued a new membership; re-serving on it`);
|
||||
for (const h of issuedHandlers) h(m);
|
||||
} catch (err) {
|
||||
console.log(`[mesh-tools] a membership arrived that is not one: ${err}`);
|
||||
}
|
||||
}
|
||||
})();
|
||||
} catch {
|
||||
// A subscription this account may not make is a mesh older than the membership.
|
||||
}
|
||||
};
|
||||
await follow(self);
|
||||
|
||||
/** The subjects a tool of a served module is served on: from that module's membership when
|
||||
* issued, derived otherwise (the shape the mesh issues on day one, so the two agree). */
|
||||
const servedOn = (module: string, tool: string): { subject: string; queue?: string }[] => {
|
||||
const m = issued.get(module);
|
||||
if (m) {
|
||||
const out = m.serves.map((s) => ({ subject: s.subject.replace("{tool}", tool), queue: s.queue }));
|
||||
// The verb that lists what this module serves is answered on the mesh's plain address for it
|
||||
// whatever the placement — one answer suffices, so a queue — and on this machine's beside it.
|
||||
if (tool === "tools" && m.tools && !out.some((s) => s.subject === m.tools)) {
|
||||
out.unshift({ subject: m.tools, queue: `serve.${module}` });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
const base = `mesh.mod.${module}.tool.${tool}`;
|
||||
const out: { subject: string; queue?: string }[] = [{ subject: base, queue: `serve.${module}` }];
|
||||
if (node) out.push({ subject: `${base}.${node}` });
|
||||
return out;
|
||||
};
|
||||
|
||||
/** Where a call by key goes: a subject this connection's own membership says it reaches, when it
|
||||
* says one — the machine's when named — else the derived shape. */
|
||||
const reachedAt = (key: string): string => {
|
||||
const [name, wanted] = key.split("@", 2);
|
||||
const reach = issued.get(self)?.reaches?.[name];
|
||||
if (reach && reach.length > 0) {
|
||||
if (wanted) {
|
||||
const at = reach.find((s) => s.endsWith(`.${wanted}`));
|
||||
if (at) return at;
|
||||
} else {
|
||||
return reach[0];
|
||||
}
|
||||
}
|
||||
return toolSubject(key, self);
|
||||
};
|
||||
|
||||
/** Answer one subject with one handler, and say which machine answered (novox/hq ADR 0159). */
|
||||
const answerOn = <Req, Res>(
|
||||
subject: string,
|
||||
queue: string | undefined,
|
||||
handler: (body: Req) => Promise<Res>,
|
||||
): (() => void) => {
|
||||
const sub = conn.subscribe(subject, queue ? { queue } : {});
|
||||
subs.push(sub);
|
||||
void (async () => {
|
||||
for await (const msg of sub) {
|
||||
let reply: { result?: Res; error?: string; node?: string };
|
||||
try {
|
||||
reply = { result: await handler(JSON.parse(sc.decode(msg.data)) as Req) };
|
||||
} catch (err) {
|
||||
// The caller is told, rather than left to time out: a handler that threw is a
|
||||
// different failure from a tool nobody serves, and only one of them is worth retrying.
|
||||
reply = { error: err instanceof Error ? err.message : String(err) };
|
||||
}
|
||||
if (node) reply.node = node;
|
||||
msg.respond(sc.encode(JSON.stringify(reply)));
|
||||
}
|
||||
})();
|
||||
return () => sub.unsubscribe();
|
||||
};
|
||||
|
||||
/**
|
||||
* Ask one question and await one answer, with the machine that gave it.
|
||||
*
|
||||
* Core NATS request/reply, not JetStream: a tool call must never be persisted (design 25 §3),
|
||||
* and a lost one is a timeout the caller already handles. The reply travels on the inbox the
|
||||
* request carries, which the responder may answer because its account has `allow_responses`
|
||||
* — one reply to a message it actually received, and nothing wider.
|
||||
*/
|
||||
const ask = async <Req, Res>(key: string, body: Req, on?: string): Promise<Answered<Res>> => {
|
||||
const msg = await conn.request(on ?? reachedAt(key), sc.encode(JSON.stringify(body)), {
|
||||
timeout: REQUEST_TIMEOUT_MS,
|
||||
});
|
||||
const reply = JSON.parse(sc.decode(msg.data)) as { result?: Res; error?: string; node?: string };
|
||||
if (reply.error) throw new Error(reply.error);
|
||||
return { result: reply.result as Res, node: reply.node };
|
||||
};
|
||||
|
||||
return {
|
||||
async request<Req, Res>(key: string, body: Req): Promise<Res> {
|
||||
return (await ask<Req, Res>(key, body)).result;
|
||||
},
|
||||
|
||||
ask,
|
||||
|
||||
/**
|
||||
* Answer a question, two ways (novox/hq ADR 0159): on the module's subject in a queue group,
|
||||
* so several machines may serve one tool and exactly one of them answers each call; and on the
|
||||
* same subject with this machine as its last token, so a caller that names the machine reaches
|
||||
* this instance and no other. A runtime that does not know its machine serves only the first,
|
||||
* which is how it always behaved.
|
||||
*/
|
||||
async handle<Req, Res>(key: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
|
||||
// A seat's verb named outright is served on the seat's subject as given, for a holder that
|
||||
// knows its role without a membership; everything else is a served module's own tool, served
|
||||
// where the mesh issued that module (ADR 0160). A key naming a module this connection does
|
||||
// not follow is not served here at all: a module serves its own tools, and the node's
|
||||
// runtime those of the modules it was given (ADR 0175) — never a stranger's.
|
||||
if (key.startsWith("seat:")) {
|
||||
const stop = answerOn(toolSubject(key, self), undefined, handler);
|
||||
return () => stop();
|
||||
}
|
||||
const dot = key.indexOf(".");
|
||||
const module = dot < 0 ? self : key.slice(0, dot);
|
||||
if (!issued.has(module) && module !== self) {
|
||||
throw new Error(
|
||||
`${self} cannot serve ${key}: a module serves its own tools, and a runtime those of the ` +
|
||||
"modules it follows",
|
||||
);
|
||||
}
|
||||
const tool = dot < 0 ? key : key.slice(dot + 1);
|
||||
let stops = servedOn(module, tool).map((s) => answerOn(s.subject, s.queue, handler));
|
||||
// When that module's new membership arrives, serve where it now says and stop serving where
|
||||
// it no longer does.
|
||||
issuedHandlers.push((m) => {
|
||||
if (m.module !== module) return;
|
||||
stops.forEach((stop) => stop());
|
||||
stops = servedOn(module, tool).map((s) => answerOn(s.subject, s.queue, handler));
|
||||
});
|
||||
return () => stops.forEach((stop) => stop());
|
||||
},
|
||||
|
||||
module: self,
|
||||
follow,
|
||||
serving: () => [...issued.keys()],
|
||||
membership: (module?: string) => issued.get(module ?? self),
|
||||
onMembership: (handler: (m: Membership) => void) => {
|
||||
issuedHandlers.push(handler);
|
||||
},
|
||||
|
||||
async handleSubject<Req, Res>(subject: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
|
||||
return answerOn(subject, undefined, handler);
|
||||
},
|
||||
|
||||
/**
|
||||
* Emit an event.
|
||||
*
|
||||
* Published into JetStream and awaited, so a publish the bus never accepted fails the emit
|
||||
* rather than vanishing — at-least-once starts at the emitter, not only the consumer
|
||||
* (ADR 0042).
|
||||
*
|
||||
* `msgID` is the event's own id, so a redelivery after a crash between publishing and
|
||||
* acknowledging is de-duplicated by the server inside its window rather than seen twice.
|
||||
*/
|
||||
async publish<T>(env: Envelope<T>): Promise<void> {
|
||||
// **The body is the payload and the metadata rides as headers** (ADR 0042). That shape
|
||||
// is what the conformance suite pins: an implementation that nested the whole envelope in
|
||||
// the body would pass every one of its own tests and agree with nobody.
|
||||
const meta = (env.headers ?? {}) as Record<string, string>;
|
||||
const h = natsHeaders();
|
||||
for (const [k, v] of Object.entries(meta)) {
|
||||
if (v != null) h.set(k, String(v));
|
||||
}
|
||||
if (!meta["content-type"]) h.set("content-type", "application/json");
|
||||
if (env.node) h.set("x-node", env.node);
|
||||
|
||||
// The emitting module's subject: the tool at work's when a served module's tool emits from
|
||||
// the node's runtime (ADR 0175), this connection's own otherwise.
|
||||
await js.publish(eventSubject(env.key, atWork.getStore()?.module ?? self), sc.encode(JSON.stringify(env.body)), {
|
||||
headers: h,
|
||||
// De-duplicated by the server inside its window, so a redelivery after a crash between
|
||||
// publishing and acknowledging is not seen twice. Only the emitter can make this id.
|
||||
msgID: meta["x-event-id"],
|
||||
});
|
||||
},
|
||||
|
||||
/**
|
||||
* React to events.
|
||||
*
|
||||
* The durable consumer is the **controller's** to create, from what this module declared it
|
||||
* consumes (design 29 §3) — this binds to it and never creates one. A runtime that created
|
||||
* its own would be a module deciding its own delivery semantics, and its account cannot
|
||||
* reach the JetStream API to do it anyway.
|
||||
*
|
||||
* **One consumer, one loop, however many patterns a module registers.** A module has exactly one
|
||||
* durable consumer, so two loops reading it would each take half the messages — and a loop that
|
||||
* received one its own pattern does not match acknowledges it, which is the right answer for a
|
||||
* filter wider than anything registered and silent loss when it is another handler's. Every
|
||||
* registration is therefore dispatched from one reader, and a message is acknowledged once every
|
||||
* handler it is for has taken it.
|
||||
*/
|
||||
async subscribe<T>(
|
||||
pattern: string,
|
||||
handler: (env: Envelope<T>) => Promise<void>,
|
||||
): Promise<() => void> {
|
||||
const listener = { pattern, handler: handler as (env: Envelope<unknown>) => Promise<void> };
|
||||
listeners.push(listener);
|
||||
if (!reading) {
|
||||
const durable = `${cred.node ?? "?"}_${self}`;
|
||||
const consumer = await js.consumers.get("EVENTS", durable);
|
||||
const messages = await consumer.consume();
|
||||
reading = messages;
|
||||
void (async () => {
|
||||
for await (const msg of messages) {
|
||||
await deliver(msg, listeners);
|
||||
}
|
||||
})();
|
||||
}
|
||||
return () => {
|
||||
const at = listeners.indexOf(listener);
|
||||
if (at >= 0) listeners.splice(at, 1);
|
||||
if (listeners.length === 0 && reading) {
|
||||
void reading.close();
|
||||
reading = undefined;
|
||||
}
|
||||
};
|
||||
},
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
for (const sub of subs) sub.unsubscribe();
|
||||
// Drain rather than close: an in-flight reply is finished instead of dropped, which for a
|
||||
// tool call is the difference between an answer and an unexplained timeout at the caller.
|
||||
await conn.drain();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver one event to every handler it is for, acknowledging only once each has taken it.
|
||||
*
|
||||
* Several registrations share one durable consumer, so matching happens here rather than by having
|
||||
* each registration read the stream: two readers of one consumer would split it between them, and a
|
||||
* message that reached the wrong one would be acknowledged as not-for-me and lost.
|
||||
*/
|
||||
async function deliver(
|
||||
msg: JsMsg,
|
||||
listeners: { pattern: string; handler: (env: Envelope<unknown>) => Promise<void> }[],
|
||||
): Promise<void> {
|
||||
let env: Envelope<unknown>;
|
||||
try {
|
||||
env = toEnvelope<unknown>(msg);
|
||||
} catch {
|
||||
// Unparseable: acknowledge it. Redelivering a message no version of this code can read is
|
||||
// an infinite loop, and the stream's dead-letter is for handlers that fail, not for bytes
|
||||
// that were never an envelope.
|
||||
msg.term();
|
||||
return;
|
||||
}
|
||||
const forThis = listeners.filter((l) => topicMatches(l.pattern, env.key));
|
||||
if (forThis.length === 0) {
|
||||
// The consumer's filters are the controller's, derived from what the module declared it
|
||||
// consumes, and may be wider than anything it registered a handler for. Acknowledge it, or it
|
||||
// would be redelivered until it expired.
|
||||
msg.ack();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
for (const l of forThis) await l.handler(env);
|
||||
msg.ack();
|
||||
} catch {
|
||||
// Negative-acknowledge with a delay, so a handler failing on a transient cause gets another
|
||||
// attempt, and one failing permanently exhausts max-deliver and dead-letters rather than
|
||||
// spinning. The consumer's limits are the controller's; this only says "not done".
|
||||
msg.nak(5_000);
|
||||
}
|
||||
}
|
||||
|
||||
/** Rebuild the envelope a module sees, from the subject, the headers and the payload — the
|
||||
* mirror of publish, and the reason both live beside each other. */
|
||||
function toEnvelope<T>(msg: JsMsg): Envelope<T> {
|
||||
const headers: Record<string, string> = {};
|
||||
if (msg.headers) {
|
||||
for (const k of msg.headers.keys()) headers[k] = msg.headers.get(k);
|
||||
}
|
||||
return {
|
||||
// The event's own key, recovered from the subject: `mesh.mod.<module>.event.<key>`. The
|
||||
// module never sees the subject, only the key it declared.
|
||||
key: keyFromSubject(msg.subject),
|
||||
node: headers["x-node"] ?? "",
|
||||
body: JSON.parse(sc.decode(msg.data)) as T,
|
||||
headers: headers as EventHeaders,
|
||||
};
|
||||
}
|
||||
|
||||
/** The event key a module sees: the emitter and the event, which is exactly how its manifest names
|
||||
* what it consumes (novox/hq design 29 §1, 04-ISSUES/127).
|
||||
*
|
||||
* **One vocabulary for the declaration and the handler.** This returned the event name alone, so a
|
||||
* manifest declaring `consumes: builder.built` produced a handler pattern that could never match
|
||||
* the key it was compared against — and a module consuming the same event from two emitters could
|
||||
* not tell them apart except by reading a header. The subject already carries the emitter; naming it
|
||||
* here makes a mismatch between manifest and code a typo rather than a category error. */
|
||||
function keyFromSubject(subject: string): string {
|
||||
const marker = ".event.";
|
||||
const at = subject.indexOf(marker);
|
||||
if (at < 0) return subject;
|
||||
const event = subject.slice(at + marker.length);
|
||||
// `mesh.mod.<emitter>.event.…` — the emitter is the token before the marker.
|
||||
const before = subject.slice(0, at).split(".");
|
||||
const emitter = before[before.length - 1];
|
||||
return emitter ? `${emitter}.${event}` : event;
|
||||
}
|
||||
|
||||
/** A module's own event subject. Derived, never taken from the caller: the module names its
|
||||
* event and the mesh decides where it lands (design 29 §1). */
|
||||
function eventSubject(type: string, self: string): string {
|
||||
return `mesh.mod.${self}.event.${type}`;
|
||||
}
|
||||
|
||||
/** A tool's subject. A bare name is this module's own tool; `<module>.<tool>` addresses
|
||||
* another's, which is how a request reaches a module that is not this one; `seat:<seat>.<verb>`
|
||||
* addresses a role's tool, answered by whoever holds the seat (novox/hq ADR 0132) — with
|
||||
* `seat:<seat>.<verb>@<node>` for a node-scoped seat, whose tool carries the machine (design 33 §4). */
|
||||
function toolSubject(key: string, self: string): string {
|
||||
if (key.startsWith("seat:")) {
|
||||
const rest = key.slice("seat:".length);
|
||||
const dot = rest.indexOf(".");
|
||||
if (dot < 0) throw new Error(`"${key}" names a seat and no verb: seat:<seat>.<verb>`);
|
||||
const seat = rest.slice(0, dot);
|
||||
const [verb, node] = rest.slice(dot + 1).split("@", 2);
|
||||
return node ? `mesh.seat.${seat}.tool.${verb}.${node}` : `mesh.seat.${seat}.tool.${verb}`;
|
||||
}
|
||||
// `<module>.<tool>@<node>` names the machine (novox/hq ADR 0159): the same subject with the
|
||||
// machine as its last token, which is what that instance serves beside the queue.
|
||||
const [name, node] = key.split("@", 2);
|
||||
const dot = name.indexOf(".");
|
||||
const base = dot < 0
|
||||
? `mesh.mod.${self}.tool.${name}`
|
||||
: `mesh.mod.${name.slice(0, dot)}.tool.${name.slice(dot + 1)}`;
|
||||
return node ? `${base}.${node}` : base;
|
||||
}
|
||||
|
||||
/** A seat's verb, as its holder serves it: flat for a mesh seat, carrying the machine for a
|
||||
* node-scoped one (design 33 §4) — the same shape the controller grants. */
|
||||
export function seatToolSubject(seat: string, verb: string, scope: string | undefined, node: string | undefined): string {
|
||||
const base = `mesh.seat.${seat}.tool.${verb}`;
|
||||
return scope === "node" && node ? `${base}.${node}` : base;
|
||||
}
|
||||
|
||||
function normalizeFingerprint(fingerprint: string): string {
|
||||
return fingerprint.replace(/^sha256:/i, "").replace(/:/g, "").toLowerCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Dial once to see the certificate, and refuse unless it is exactly the one the mesh pinned.
|
||||
* A certificate authority is not consulted: the mesh issued this and knows its fingerprint,
|
||||
* which is stronger than trusting whoever a machine's trust store happens to contain.
|
||||
*
|
||||
* **The pin is the only check.** What comes back is handed to the client as its TLS options, and
|
||||
* the client's transport spreads them into Node's own `tls.connect` — so the pinned certificate
|
||||
* is the one authority the handshake accepts, and the hostname check beside it is replaced with
|
||||
* one that always passes. Pinning the exact certificate makes verifying its name redundant, and
|
||||
* the bus's certificate names the seat (`mesh-broker`), not the address a machine happens to
|
||||
* dial it by: every module on the mesh met "does not match certificate's altnames" the first time
|
||||
* it reached the handshake (2026-09-28).
|
||||
*/
|
||||
async function pinnedTls(rawUrl: string, fingerprint: string): Promise<TlsOptions> {
|
||||
const url = new URL(rawUrl.includes("://") ? rawUrl : `nats://${rawUrl}`);
|
||||
const port = url.port ? Number(url.port) : 4222;
|
||||
// **The bus speaks first, in the clear.** A NATS server sends its INFO line before TLS begins,
|
||||
// and only then expects the client to start the handshake; a raw TLS connect to that port reads
|
||||
// the INFO line as a TLS record and fails with "wrong version number" — which is what every
|
||||
// module met the first time it dialled the bus being built (2026-09-28). So: connect, wait for
|
||||
// INFO, then start TLS on the same socket, and read the certificate the server presents.
|
||||
const certificate = await new Promise<tls.DetailedPeerCertificate>((resolve, reject) => {
|
||||
const plain = net.connect({ host: url.hostname, port }, () => {});
|
||||
let seenInfo = false;
|
||||
let buffered = "";
|
||||
plain.on("error", reject);
|
||||
plain.on("data", (chunk: Buffer) => {
|
||||
if (seenInfo) return;
|
||||
buffered += chunk.toString("utf8");
|
||||
if (!buffered.includes("\r\n")) return;
|
||||
seenInfo = true;
|
||||
plain.removeAllListeners("data");
|
||||
const secure = tls.connect(
|
||||
{ socket: plain, rejectUnauthorized: false, servername: url.hostname },
|
||||
() => {
|
||||
const peer = secure.getPeerCertificate(true);
|
||||
secure.end();
|
||||
resolve(peer);
|
||||
},
|
||||
);
|
||||
secure.on("error", reject);
|
||||
});
|
||||
});
|
||||
const seen = createHash("sha256").update(certificate.raw).digest("hex");
|
||||
if (seen !== normalizeFingerprint(fingerprint)) {
|
||||
throw new PinMismatchError(
|
||||
`the bus at ${url.hostname}:${port} presented ${seen}, not the pinned ${normalizeFingerprint(fingerprint)}`,
|
||||
);
|
||||
}
|
||||
const pem = `-----BEGIN CERTIFICATE-----\n${certificate.raw.toString("base64").replace(/(.{64})/g, "$1\n")}\n-----END CERTIFICATE-----\n`;
|
||||
// Node's option, not the client's: the transport passes the whole object on. `undefined` from
|
||||
// checkServerIdentity is "the name is fine"; the pin above already decided the rest.
|
||||
return { ca: pem, checkServerIdentity: () => undefined } as TlsOptions;
|
||||
}
|
||||
|
||||
/** The mesh's topic matching: `*` is one token, `#` the rest. This is the module's vocabulary —
|
||||
* a module's `consumes` pattern is matched here, and the subject it becomes is the mesh's
|
||||
* business, not the module's. */
|
||||
export function topicMatches(pattern: string, key: string): boolean {
|
||||
return matchFrom(pattern.split("."), 0, key.split("."), 0);
|
||||
}
|
||||
|
||||
function matchFrom(p: string[], pi: number, k: string[], ki: number): boolean {
|
||||
if (pi === p.length) return ki === k.length;
|
||||
// `**` is the mesh's wildcard for the rest of a name; `#` is the old bus's, accepted so a pattern
|
||||
// written either way behaves the same while both buses ship (novox/hq design 29 §1).
|
||||
if (p[pi] === "#" || p[pi] === "**") {
|
||||
for (let skip = ki; skip <= k.length; skip++) {
|
||||
if (matchFrom(p, pi + 1, k, skip)) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
if (ki === k.length) return false;
|
||||
if (p[pi] !== "*" && p[pi] !== k[ki]) return false;
|
||||
return matchFrom(p, pi + 1, k, ki + 1);
|
||||
}
|
||||
@@ -0,0 +1,280 @@
|
||||
/**
|
||||
* The mesh's tools, for whoever is on a machine (novox/hq design 25 §7, design 34).
|
||||
*
|
||||
* Two surfaces over one thing. A command line, for somebody at a terminal; an MCP server, for an
|
||||
* agent. Both are adapters over the same three calls — what tools are there, what does this one take,
|
||||
* call it — because a second way of reaching a tool is a second thing to keep correct.
|
||||
*
|
||||
* **It uses the same client a module's runtime uses.** Not a second protocol and not a bridge: the
|
||||
* caller connects as its own bus user — a person's, or the console's — publishes on the tool subjects
|
||||
* that account permits, and the server refuses anything else. So "what may this ask" is answered by the
|
||||
* same permission list that answers it for a module, and there is nothing here for an audit to read
|
||||
* separately.
|
||||
*
|
||||
* What the caller may NOT do is the more interesting half, and none of it is enforced here — it is the
|
||||
* account (design 25 §4): it cannot publish an event, so it cannot claim a module said something; it
|
||||
* has no consumer, so there is no delivery to acknowledge; and it cannot answer a request, so it cannot
|
||||
* impersonate a module on a bus where anyone may serve a tool.
|
||||
*/
|
||||
import { readFile } from "node:fs/promises";
|
||||
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
import { connectNats, type Answered, type Credential } from "./broker-nats.js";
|
||||
import { TOOLS_VERB, type ToolsAnswer } from "./runtime.js";
|
||||
|
||||
/** Where the catalogue answers which modules the mesh holds. */
|
||||
const CATALOGUE_MODULES = "mesh-catalog.catalog_modules";
|
||||
|
||||
/** Where the mesh answers every role's tools, from its records: the mesh-controller seat's own
|
||||
* `tools` verb (novox/hq ADR 0154, design 33 §5). */
|
||||
const SEAT_TOOLS = "seat:mesh-controller.tools";
|
||||
|
||||
/** A tool as its module describes it. */
|
||||
export interface Tool {
|
||||
/** The module that serves it — or, for a role's tool, the seat. */
|
||||
module: string;
|
||||
name: string;
|
||||
description?: string;
|
||||
/** The JSON schema of what it takes, as the module declared it. */
|
||||
input?: unknown;
|
||||
/** True for a role's tool: addressed to the seat, answered by whoever holds it (ADR 0132). */
|
||||
seat?: boolean;
|
||||
/** A role's scope: `node` for a seat held once per machine, whose verb is asked of one machine
|
||||
* (design 33 §4) and takes `node` for it; `mesh` or absent otherwise. */
|
||||
scope?: string;
|
||||
/** Where the tool is answered, as the mesh issued it (ADR 0160): the plain subject first when the
|
||||
* module answers for itself anywhere, then one per machine. Absent for a runtime older than this. */
|
||||
subjects?: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* What the mesh could say about its tools when asked (design 34 §3).
|
||||
*
|
||||
* **Silence is named, never dropped.** A module the catalogue holds and nothing answered for is in
|
||||
* `notAnswering`, because a tool that is not offered looks exactly like a tool that does not exist,
|
||||
* and those need different people to fix them.
|
||||
*/
|
||||
export interface Listing {
|
||||
tools: Tool[];
|
||||
/** Modules the catalogue holds whose runtime did not answer `tools`: not assigned, not up, or built
|
||||
* before the runtime answered it. Each may still be called by name. The mesh's own records are
|
||||
* listed here as `mesh-controller (seat)` when the control plane did not answer. */
|
||||
notAnswering: string[];
|
||||
}
|
||||
|
||||
/** The seats and the verbs each declares, from the last listing, so a call can tell a role's tool
|
||||
* from a module's when the two share a prefix (a module and a seat may share a name). */
|
||||
export type Seats = Map<string, Set<string>>;
|
||||
|
||||
/** The roles' tools, keyed the way `toolKey` names them. */
|
||||
export function seatsIn(have: Listing): Seats {
|
||||
const seats: Seats = new Map();
|
||||
for (const t of have.tools) {
|
||||
if (!t.seat) continue;
|
||||
if (!seats.has(t.module)) seats.set(t.module, new Set());
|
||||
seats.get(t.module)!.add(t.name);
|
||||
}
|
||||
return seats;
|
||||
}
|
||||
|
||||
/** The key a call uses for `<prefix>.<name>`: a role's when the prefix is a seat declaring that
|
||||
* verb, a module's otherwise. Both names for one capability are deliberate and bounded (ADR 0132);
|
||||
* the seat wins only for a verb it actually declares, so a module's own tool is never shadowed. */
|
||||
export function toolKey(name: string, seats?: Seats): string {
|
||||
if (name.startsWith("seat:")) return name;
|
||||
const dot = name.indexOf(".");
|
||||
if (dot < 0) return name;
|
||||
const prefix = name.slice(0, dot);
|
||||
const verb = name.slice(dot + 1);
|
||||
if (seats?.get(prefix)?.has(verb)) return `seat:${prefix}.${verb}`;
|
||||
return name;
|
||||
}
|
||||
|
||||
/**
|
||||
* A person's credential, as `operator issue` prints it.
|
||||
*
|
||||
* The same shape a module is handed, minus the parts a module needs and a person does not: no node,
|
||||
* because a person is not on a machine, and no module, because they are not one.
|
||||
*/
|
||||
export interface PersonCredential extends Credential {
|
||||
person?: string;
|
||||
invokes?: string[];
|
||||
}
|
||||
|
||||
/** Read the credential from the file `operator issue` produced. */
|
||||
export async function credentialFrom(path: string): Promise<PersonCredential> {
|
||||
const raw = await readFile(path, "utf8");
|
||||
let held: PersonCredential;
|
||||
try {
|
||||
held = JSON.parse(raw) as PersonCredential;
|
||||
} catch (e) {
|
||||
throw new Error(
|
||||
`${path} is not a credential this mesh issued: ${(e as Error).message}. ` +
|
||||
"It is the JSON `operator issue` printed, saved verbatim.",
|
||||
);
|
||||
}
|
||||
if (!held.url || !held.user || !held.password) {
|
||||
throw new Error(
|
||||
`${path} names no bus, user or password. It is the JSON \`operator issue\` printed, saved ` +
|
||||
"verbatim — not an edited copy of it.",
|
||||
);
|
||||
}
|
||||
return held;
|
||||
}
|
||||
|
||||
/** Connect as this person. The module name the runtime wants is their own user, because every subject
|
||||
* it derives is for a tool somebody else serves. */
|
||||
export async function connectAs(held: PersonCredential): Promise<Broker> {
|
||||
return connectNats({ ...held, module: held.user });
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect as the console: the module credential the mesh delivered (novox/hq ADR 0152), read from
|
||||
* the same variable every runtime reads. It names the node and the module, so the account's inbox
|
||||
* and subjects derive from what the mesh authorised and from nothing in this process's environment.
|
||||
*/
|
||||
export async function connectAsTheConsole(path: string): Promise<{ bus: Broker; who: string }> {
|
||||
const raw = await readFile(path, "utf8");
|
||||
let held: Credential;
|
||||
try {
|
||||
held = JSON.parse(raw) as Credential;
|
||||
} catch (e) {
|
||||
throw new Error(`${path} is not a broker credential: ${(e as Error).message}`);
|
||||
}
|
||||
if (!held.url || !held.module || !held.user) {
|
||||
throw new Error(
|
||||
`${path} names no bus, module or user: the console runs on the credential the mesh sealed to ` +
|
||||
"this machine for it, and nothing else",
|
||||
);
|
||||
}
|
||||
return { bus: await connectNats(held), who: `${held.node ?? "?"}.${held.module}` };
|
||||
}
|
||||
|
||||
/**
|
||||
* What tools the mesh has, asked of the modules (design 34 §3).
|
||||
*
|
||||
* The catalogue says which modules the mesh holds; each module says what it serves, through the one
|
||||
* verb its runtime answers for it. **Asked, not configured**: a client carrying its own list would be a
|
||||
* list that goes stale the first time a module is assigned. Every module is asked at once, and the bus
|
||||
* refuses at once a request nothing serves, so the cost is bounded by the modules that are up.
|
||||
*/
|
||||
export async function toolsOn(bus: Broker): Promise<Listing> {
|
||||
const [answered, roles] = await Promise.all([
|
||||
bus.request<Record<string, never>, { modules?: { module: string }[] }>(CATALOGUE_MODULES, {}),
|
||||
// The roles' tools, from the mesh's records (design 33 §5). Asked beside the modules rather
|
||||
// than first: a control plane that is restarting must not hide every module's tools with it.
|
||||
bus
|
||||
.request<Record<string, never>, { seats?: { seat: string; scope?: string; tools?: ToolsAnswer["tools"] }[] }>(
|
||||
SEAT_TOOLS,
|
||||
{},
|
||||
)
|
||||
.catch(() => undefined),
|
||||
]);
|
||||
const names = (answered.modules ?? []).map((m) => m.module).filter((m) => typeof m === "string");
|
||||
|
||||
const asked = await Promise.allSettled(
|
||||
names.map((module) => bus.request<Record<string, never>, ToolsAnswer>(`${module}.${TOOLS_VERB}`, {})),
|
||||
);
|
||||
const tools: Tool[] = [];
|
||||
const notAnswering: string[] = [];
|
||||
if (roles) {
|
||||
for (const s of roles.seats ?? []) {
|
||||
// A node-scoped seat's tool is asked of one machine (design 33 §4): listed with its scope, so
|
||||
// a caller names the machine and the call carries it — `seat:<seat>.<verb>@<node>`. Left out
|
||||
// of the listing, the verb never resolved as a seat's and nothing served it (ADR 0170).
|
||||
for (const t of s.tools ?? []) {
|
||||
tools.push({ module: s.seat, name: t.name, description: t.description, input: t.input, seat: true, scope: s.scope });
|
||||
}
|
||||
}
|
||||
} else {
|
||||
notAnswering.push("mesh-controller (seat)");
|
||||
}
|
||||
asked.forEach((outcome, i) => {
|
||||
const module = names[i]!;
|
||||
if (outcome.status === "fulfilled" && typeof outcome.value?.failed === "string") {
|
||||
// The runtime answered for it and serves nothing: the bundle failed to load (ADR 0175). Said
|
||||
// with the reason, because "not answering" would send somebody to check an assignment that
|
||||
// is fine.
|
||||
notAnswering.push(`${module} (its tools bundle failed to load: ${outcome.value.failed})`);
|
||||
} else if (outcome.status === "fulfilled" && Array.isArray(outcome.value?.tools)) {
|
||||
for (const t of outcome.value.tools) {
|
||||
tools.push({ module, name: t.name, description: t.description, input: t.input, subjects: t.subjects });
|
||||
}
|
||||
} else {
|
||||
notAnswering.push(module);
|
||||
}
|
||||
});
|
||||
tools.sort((a, b) => `${a.module}.${a.name}`.localeCompare(`${b.module}.${b.name}`));
|
||||
notAnswering.sort();
|
||||
return { tools, notAnswering };
|
||||
}
|
||||
|
||||
/** Call one tool. The key is `<module>.<tool>`, which is what a person types and what the account
|
||||
* permits — one vocabulary, so a refusal names the thing they asked for. */
|
||||
/** Call a tool and learn which machine answered (novox/hq ADR 0159). `<module>.<tool>@<node>` asks
|
||||
* the instance on one machine; without it, whichever instance answers first does, and the answer
|
||||
* says which. */
|
||||
export async function callTool(
|
||||
bus: Broker,
|
||||
key: string,
|
||||
args: unknown,
|
||||
seats?: Seats,
|
||||
listing?: Listing,
|
||||
): Promise<Answered<unknown>> {
|
||||
const at = key.indexOf("@");
|
||||
const name = at < 0 ? key : key.slice(0, at);
|
||||
const node = at < 0 ? "" : key.slice(at + 1);
|
||||
if (!name.includes(".")) {
|
||||
throw new Error(
|
||||
`"${key}" does not name a tool: write <module>.<tool>, as \`mesh tools\` lists them, ` +
|
||||
"or <module>.<tool>@<node> for the instance on one machine",
|
||||
);
|
||||
}
|
||||
const resolved = toolKey(name, seats) + (node ? `@${node}` : "");
|
||||
// Where the tool is answered is the module's to say and the mesh's to issue (ADR 0160): when the
|
||||
// listing carried subjects for it, the call goes to one of those and composes nothing.
|
||||
const on = subjectListed(name, node, listing);
|
||||
const asking = bus as Broker & { ask?: <Req, Res>(k: string, b: Req, on?: string) => Promise<Answered<Res>> };
|
||||
if (typeof asking.ask === "function") return asking.ask<unknown, unknown>(resolved, args ?? {}, on);
|
||||
return { result: await bus.request<unknown, unknown>(resolved, args ?? {}) };
|
||||
}
|
||||
|
||||
/** The subject the listing says answers `<module>.<tool>` — the machine's when one is named, else
|
||||
* the plain one — or undefined when the listing said none, and the key is composed as before. */
|
||||
export function subjectListed(name: string, node: string, listing?: Listing): string | undefined {
|
||||
if (!listing) return undefined;
|
||||
const dot = name.indexOf(".");
|
||||
const module = name.slice(0, dot);
|
||||
const tool = name.slice(dot + 1);
|
||||
const found = listing.tools.find((t) => t.module === module && t.name === tool && !t.seat);
|
||||
const subjects = found?.subjects ?? [];
|
||||
if (subjects.length === 0) return undefined;
|
||||
if (node) return subjects.find((s) => s.endsWith(`.${node}`));
|
||||
return subjects[0];
|
||||
}
|
||||
|
||||
/**
|
||||
* Why a call failed, said so that the remedy is in the words.
|
||||
*
|
||||
* Three answers a person actually gets, and they need different things done: nobody serves that tool,
|
||||
* the mesh refused this account, or the tool itself failed. Without this they are one timeout and a
|
||||
* stack trace.
|
||||
*/
|
||||
export function whyItFailed(key: string, err: unknown): string {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
if (/no responders|503/i.test(message)) {
|
||||
return `nothing serves ${key}. The module may not be assigned to any machine, or it is down` +
|
||||
(key.startsWith("seat:") ? ", or nothing holds that seat" : "") +
|
||||
" — `mesh tools` lists what answered.";
|
||||
}
|
||||
if (/permissions violation|authorization/i.test(message)) {
|
||||
return `this account may not call ${key}. What it may call was fixed when it was issued — a ` +
|
||||
"person's by `operator issue`, the console's by its manifest.";
|
||||
}
|
||||
if (/timeout/i.test(message)) {
|
||||
return `${key} did not answer in time. Something is serving it, so this is the tool being slow ` +
|
||||
"rather than absent.";
|
||||
}
|
||||
return `${key} failed: ${message}`;
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
/**
|
||||
* The console's endpoint: MCP over HTTP, on a machine's loopback (novox/hq ADR 0152, design 34 §2).
|
||||
*
|
||||
* **Loopback is the authority boundary.** Whoever can connect is on the machine, and whoever is on the
|
||||
* machine is the account that owns the mesh there (ADR 0034, ADR 0144). So there is no token and no
|
||||
* login here, and the one thing this file enforces is that it binds nothing else: a console reachable
|
||||
* from another machine would be authority over the mesh handed to whoever finds the port.
|
||||
*
|
||||
* The transport is the streamable-HTTP shape an agent host speaks: `POST /mcp` with one JSON-RPC
|
||||
* message, answered with one JSON body. No session, because the surface holds nothing per caller; no
|
||||
* event stream, because nothing here has anything to say unasked.
|
||||
*/
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
|
||||
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
import { mcpSurface, type Reply, type Request } from "./mcp.js";
|
||||
|
||||
/** The most a request body may be. A tool's arguments are small; a megabyte is somebody else's file. */
|
||||
const BODY_LIMIT = 1 << 20;
|
||||
|
||||
export interface Listening {
|
||||
/** Where it listens, as `host:port`, with the port the machine actually gave. */
|
||||
address: string;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
/** Hosts that are this machine and no other. */
|
||||
const loopback = new Set(["127.0.0.1", "::1", "localhost", "[::1]"]);
|
||||
|
||||
/**
|
||||
* Listen on `host:port`. Refused unless the host is loopback — said before binding, so a manifest or
|
||||
* a flag that would open the console to a network is a startup failure rather than something
|
||||
* discovered by whoever finds it.
|
||||
*/
|
||||
export async function serveMcpHttp(bus: Broker, who: string, listen: string): Promise<Listening> {
|
||||
const at = listen.lastIndexOf(":");
|
||||
if (at < 0) {
|
||||
throw new Error(`"${listen}" is not host:port`);
|
||||
}
|
||||
const host = listen.slice(0, at);
|
||||
const port = Number(listen.slice(at + 1));
|
||||
if (!loopback.has(host)) {
|
||||
throw new Error(
|
||||
`the console listens on loopback and nowhere else (novox/hq ADR 0152): "${host}" is not this ` +
|
||||
"machine's own address — whoever is on the machine owns the mesh there, and nobody else may reach this",
|
||||
);
|
||||
}
|
||||
if (!Number.isInteger(port) || port < 0 || port > 65535) {
|
||||
throw new Error(`"${listen.slice(at + 1)}" is not a port`);
|
||||
}
|
||||
|
||||
const surface = mcpSurface(bus, who);
|
||||
const server = createServer((req, res) => {
|
||||
void route(req, res, surface.handle).catch((e) => {
|
||||
json(res, 500, { jsonrpc: "2.0", id: null, error: { code: -32603, message: String(e) } });
|
||||
});
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(port, host.replace(/^\[|\]$/g, ""), () => resolve());
|
||||
});
|
||||
const bound = server.address();
|
||||
const address = typeof bound === "object" && bound ? `${host}:${bound.port}` : listen;
|
||||
return {
|
||||
address,
|
||||
close: () =>
|
||||
new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
async function route(
|
||||
req: IncomingMessage,
|
||||
res: ServerResponse,
|
||||
handle: (r: Request) => Promise<Reply | undefined>,
|
||||
): Promise<void> {
|
||||
const path = (req.url ?? "/").split("?")[0];
|
||||
if (path === "/") {
|
||||
res.writeHead(200, { "content-type": "text/plain; charset=utf-8" });
|
||||
res.end("the mesh's console: MCP over HTTP at POST /mcp (novox/hq design 34)\n");
|
||||
return;
|
||||
}
|
||||
if (path !== "/mcp") {
|
||||
json(res, 404, { error: "the console serves /mcp and nothing else" });
|
||||
return;
|
||||
}
|
||||
switch (req.method) {
|
||||
case "POST":
|
||||
break;
|
||||
case "DELETE":
|
||||
// A host ending a session. There is no session to end; saying so is the truthful answer.
|
||||
res.writeHead(204).end();
|
||||
return;
|
||||
case "GET":
|
||||
// A host opening an event stream. The console has nothing to say unasked.
|
||||
res.writeHead(405, { allow: "POST, DELETE" }).end();
|
||||
return;
|
||||
default:
|
||||
res.writeHead(405, { allow: "POST, DELETE" }).end();
|
||||
return;
|
||||
}
|
||||
|
||||
let body: string;
|
||||
try {
|
||||
body = await read(req);
|
||||
} catch (e) {
|
||||
json(res, 413, { jsonrpc: "2.0", id: null, error: { code: -32600, message: String(e) } });
|
||||
return;
|
||||
}
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(body);
|
||||
} catch {
|
||||
json(res, 400, { jsonrpc: "2.0", id: null, error: { code: -32700, message: "the body is not JSON" } });
|
||||
return;
|
||||
}
|
||||
|
||||
// One message, or a batch of them; a batch is answered as a batch. A notification gets no reply
|
||||
// and, alone, no body: 202 is how the transport says "heard".
|
||||
if (Array.isArray(parsed)) {
|
||||
const replies = (await Promise.all(parsed.map((r) => handle(r as Request)))).filter(Boolean);
|
||||
if (replies.length === 0) {
|
||||
res.writeHead(202).end();
|
||||
} else {
|
||||
json(res, 200, replies);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const reply = await handle(parsed as Request);
|
||||
if (!reply) {
|
||||
res.writeHead(202).end();
|
||||
return;
|
||||
}
|
||||
json(res, 200, reply);
|
||||
}
|
||||
|
||||
function read(req: IncomingMessage): Promise<string> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let size = 0;
|
||||
const chunks: Buffer[] = [];
|
||||
req.on("data", (chunk: Buffer) => {
|
||||
size += chunk.length;
|
||||
if (size > BODY_LIMIT) {
|
||||
reject(new Error(`the request is larger than ${BODY_LIMIT} bytes`));
|
||||
req.destroy();
|
||||
return;
|
||||
}
|
||||
chunks.push(chunk);
|
||||
});
|
||||
req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
|
||||
req.on("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
function json(res: ServerResponse, status: number, body: unknown): void {
|
||||
const text = JSON.stringify(body);
|
||||
res.writeHead(status, {
|
||||
"content-type": "application/json; charset=utf-8",
|
||||
"content-length": Buffer.byteLength(text),
|
||||
});
|
||||
res.end(text);
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
// A tools bundle as a process the runtime launches (novox/hq ADR 0188).
|
||||
//
|
||||
// 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;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,307 @@
|
||||
// The runnable entrypoint. Three modes:
|
||||
//
|
||||
// mesh-tools serve — bind the broker and serve the assigned modules until
|
||||
// stopped; as the node-tools module, also the console on loopback. A module entrypoint that subscribes to events (on("#"))
|
||||
// starts consuming as it is imported, so this also runs consumers.
|
||||
// mesh-tools emit TYPE [JSON] emit one event onto the mesh and exit — an operable primitive,
|
||||
// and what an events test uses to put a message on the wire.
|
||||
// mesh-tools prepare bring this module's state to the shape this version needs and exit
|
||||
// — the runtime's answer to the word the mesh asks every module
|
||||
// (novox/hq ADR 0135). The entrypoints come from MESH_PREPARE, which
|
||||
// the module's own image names beside MESH_TOOL_MODULES.
|
||||
// mesh-tools run ENTRYPOINT run one compiled module entrypoint to completion and exit — the
|
||||
// runtime side of a run-once step (novox/hq ADR 0052). It imports
|
||||
// the given entrypoint, whose top-level code does its work — seed a
|
||||
// store, migrate, health-gate — and awaits it. It does NOT connect
|
||||
// to the broker: a first-boot step runs offline, before the module
|
||||
// has anything to talk to, and the host gates the container that
|
||||
// depends on it on this process exiting 0.
|
||||
//
|
||||
// The broker, in order of preference:
|
||||
// MESH_BROKER_FILE a sealed {url, fingerprint} the mesh delivered (novox/hq ADR 0043) — an
|
||||
// amqps account scoped to this module. Preferred: a module holds its own.
|
||||
// MESH_BROKER_URL a plain URL, for the bootstrap/admin case before a module has an account.
|
||||
// MESH_TOOL_MODULES <module>=/path/a,… the modules to serve and their compiled entrypoints (serve
|
||||
// mode; ADR 0175); a bare path is an entrypoint of the credential's own module
|
||||
// MESH_PREPARE /path/a,/path/b,… compiled entrypoints that prepare this module's state
|
||||
// MESH_MODULE / MESH_NODE the identity stamped onto emitted events (ADR 0042)
|
||||
|
||||
import { readFileSync } from "node:fs";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { connectNats, fatalBrokerReason as fatalNatsReason, type Credential } from "./broker-nats.js";
|
||||
import { runTools, type ServedModule } from "./runtime.js";
|
||||
import { serveMcpHttp, type Listening } from "./http.js";
|
||||
|
||||
/** The credential this process connected with, for what it says beyond the connection (ADR 0159). */
|
||||
let lastCredential: Credential | undefined;
|
||||
import { invokeTool } from "@novox/mesh-sdk/tools";
|
||||
import { useBroker } from "@novox/mesh-sdk/messaging";
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
import { emit } from "@novox/mesh-sdk/events";
|
||||
|
||||
/**
|
||||
* Connect the way this process is meant to: with its sealed credential if the mesh gave it one, and
|
||||
* over the plain bootstrap URL otherwise. A scoped module assumes the foundation's exchanges exist —
|
||||
* its account may not declare them (ADR 0043).
|
||||
*/
|
||||
// fatalBrokerReasonFor is the reason a connection failure is final rather than "not yet", for
|
||||
// whichever bus this runtime is on — each transport knows its own refusals.
|
||||
function fatalBrokerReasonFor(err: unknown): string | null {
|
||||
return fatalNatsReason(err);
|
||||
}
|
||||
|
||||
async function connectBroker(): Promise<Broker> {
|
||||
const file = process.env.MESH_BROKER_FILE;
|
||||
if (file) {
|
||||
let credential: Credential;
|
||||
try {
|
||||
credential = JSON.parse(readFileSync(file, "utf8")) as Credential;
|
||||
lastCredential = credential;
|
||||
} catch (err) {
|
||||
console.error(`mesh-tools: cannot read the broker credential at ${file}: ${err}`);
|
||||
process.exit(1);
|
||||
}
|
||||
if (!credential.url) {
|
||||
console.error(`mesh-tools: ${file} carries no url — it is not a broker credential`);
|
||||
process.exit(1);
|
||||
}
|
||||
// The mesh scoped this account to a node and module; take the runtime's identity from the
|
||||
// credential so its queue and the events it emits match what the mesh authorised, no matter
|
||||
// what the environment says.
|
||||
if (credential.node) process.env.MESH_NODE = credential.node;
|
||||
if (credential.module) process.env.MESH_MODULE = credential.module;
|
||||
// **The credential names the bus.** A module moved to the bus being built was handed a
|
||||
// credential for it — `nats://…` with user, password and fingerprint beside the address — and
|
||||
// nothing else in its environment changed (design 25; novox/hq design 28 task 5.2). The scheme
|
||||
// is enough to know which bus to speak; a runtime that always dialled the old one would keep
|
||||
// serving and answer nobody.
|
||||
// One bus (novox/hq ADR 0131, design 28 task 5.5): the credential names it, and it is this.
|
||||
return connectNats(credential);
|
||||
}
|
||||
const url = process.env.MESH_BROKER_URL;
|
||||
if (!url) {
|
||||
console.error(
|
||||
"mesh-tools: set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach",
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
return connectNats({ url });
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect for serve mode, retrying while the broker is merely not reachable yet. At startup that
|
||||
* is the NORMAL case, not a failure: a container comes up in seconds and the overlay tunnel a
|
||||
* moment later (novox/hq issue 058). Exiting instead delegated the retry to the container
|
||||
* runtime, which read as a crash-loop to every restart-counting health check and every person
|
||||
* watching. Retried indefinitely, aloud: the dependency appears or somebody reads why not.
|
||||
*
|
||||
* A failure that waiting cannot fix (see fatalBrokerReason) is thrown at once rather than retried —
|
||||
* a permanent fault masquerading as "not reachable yet" is the silent non-progress this whole
|
||||
* change exists to remove. Missing-file and empty-URL configuration errors exit inside
|
||||
* connectBroker before they reach here; a malformed URL and a refused login are caught here.
|
||||
*/
|
||||
async function connectBrokerPatiently(): Promise<Broker> {
|
||||
for (let delay = 2_000; ; delay = Math.min(delay * 2, 30_000)) {
|
||||
try {
|
||||
return await connectBroker();
|
||||
} catch (err) {
|
||||
const fatal = fatalBrokerReasonFor(err);
|
||||
if (fatal !== null) {
|
||||
console.error(`mesh-tools: ${fatal} — waiting will not fix this; giving up`);
|
||||
throw err;
|
||||
}
|
||||
const why = err instanceof Error ? err.message : String(err);
|
||||
// A little jitter so every module that was up when the broker bounced does not retry in
|
||||
// lockstep and stampede it as it recovers.
|
||||
const wait = delay + Math.floor(Math.random() * 1_000);
|
||||
console.error(`mesh-tools: the broker is not reachable yet (${why}); retrying in ${Math.round(wait / 1000)}s`);
|
||||
await new Promise((r) => setTimeout(r, wait));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* What MESH_TOOL_MODULES names (novox/hq ADR 0175, to-be 38 WP1): `<module>=<entrypoint>` entries,
|
||||
* comma-separated, several per module allowed — the node's runtime serving every assigned module's
|
||||
* bundle. A bare path is the one-module form the per-module containers still set: an entrypoint of
|
||||
* the credential's own module. Both may appear; the result is one list of modules.
|
||||
*/
|
||||
export function servedModulesFrom(spec: string, own: string | undefined): { serves: ServedModule[]; moduleEntrypoints: string[] } {
|
||||
const serves = new Map<string, string[]>();
|
||||
const moduleEntrypoints: string[] = [];
|
||||
for (const raw of spec.split(",")) {
|
||||
const entry = raw.trim();
|
||||
if (!entry) continue;
|
||||
const eq = entry.indexOf("=");
|
||||
if (eq < 0) {
|
||||
moduleEntrypoints.push(entry);
|
||||
continue;
|
||||
}
|
||||
const module = entry.slice(0, eq).trim();
|
||||
const path = entry.slice(eq + 1).trim();
|
||||
if (!module || !path) {
|
||||
throw new Error(`MESH_TOOL_MODULES: "${entry}" is neither <module>=<entrypoint> nor an entrypoint of this module`);
|
||||
}
|
||||
if (module === own) {
|
||||
moduleEntrypoints.push(path);
|
||||
continue;
|
||||
}
|
||||
serves.set(module, [...(serves.get(module) ?? []), path]);
|
||||
}
|
||||
return { serves: [...serves].map(([module, entrypoints]) => ({ module, entrypoints })), moduleEntrypoints };
|
||||
}
|
||||
|
||||
/** The module that is the node's tool runtime (novox/hq ADR 0175, to-be 38 WP3): on its credential,
|
||||
* `serve` is also the console — MCP on the machine's loopback (design 34). */
|
||||
export const RUNTIME_MODULE = "node-tools";
|
||||
|
||||
/** Where the console listens when the runtime is node-tools and nothing says otherwise: the port
|
||||
* the module's manifest declares `from: machine`. MESH_CONSOLE_LISTEN overrides it either way. */
|
||||
const CONSOLE_LISTEN = "127.0.0.1:4270";
|
||||
|
||||
async function serve(): Promise<void> {
|
||||
const broker = await connectBrokerPatiently();
|
||||
// Parsed after connecting: a bare entrypoint belongs to the module the credential names.
|
||||
const { serves, moduleEntrypoints } = servedModulesFrom(process.env.MESH_TOOL_MODULES ?? "", lastCredential?.module);
|
||||
const stop = await runTools({ broker, serves, moduleEntrypoints, credential: lastCredential });
|
||||
|
||||
// The console is this runtime's serving mode (ADR 0175 §6): as node-tools, or wherever the
|
||||
// listen address is given, the same process answers MCP on loopback for whoever is on the
|
||||
// machine. A module's own runtime in a container on the machine's network does not — two of
|
||||
// them on one port would be the fault, and the console is one per machine.
|
||||
const listen = process.env.MESH_CONSOLE_LISTEN ?? (lastCredential?.module === RUNTIME_MODULE ? CONSOLE_LISTEN : "");
|
||||
let consoleUp: Listening | undefined;
|
||||
if (listen) {
|
||||
const who = `${lastCredential?.node ?? "?"}.${lastCredential?.module ?? RUNTIME_MODULE}`;
|
||||
consoleUp = await serveMcpHttp(broker, who, listen);
|
||||
console.log(`mesh console listening on http://${consoleUp.address}/mcp as ${who}`);
|
||||
}
|
||||
|
||||
const shutdown = async (): Promise<void> => {
|
||||
stop();
|
||||
await consoleUp?.close();
|
||||
await broker.close();
|
||||
process.exit(0);
|
||||
};
|
||||
process.on("SIGTERM", () => void shutdown());
|
||||
process.on("SIGINT", () => void shutdown());
|
||||
}
|
||||
|
||||
async function emitOnce(type: string, bodyJson: string): Promise<void> {
|
||||
let body: unknown = {};
|
||||
if (bodyJson) {
|
||||
try {
|
||||
body = JSON.parse(bodyJson);
|
||||
} catch {
|
||||
console.error(`mesh-tools emit: body is not JSON: ${bodyJson}`);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
const broker = await connectBroker();
|
||||
useBroker(() => broker);
|
||||
// emit awaits the broker's publish confirm (ADR 0042), so the event is accepted before we close.
|
||||
await emit(type, body);
|
||||
await broker.close();
|
||||
}
|
||||
|
||||
async function invokeOnce(module: string, tool: string, argsJson: string): Promise<void> {
|
||||
let args: Record<string, unknown> = {};
|
||||
if (argsJson) {
|
||||
try {
|
||||
args = JSON.parse(argsJson) as Record<string, unknown>;
|
||||
} catch {
|
||||
console.error(`mesh-tools invoke: args are not JSON: ${argsJson}`);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
const broker = await connectBroker();
|
||||
const result = await invokeTool(broker, module, tool, args);
|
||||
process.stdout.write(JSON.stringify(result) + "\n");
|
||||
await broker.close();
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one compiled module entrypoint to completion — the runtime side of a run-once step
|
||||
* (novox/hq ADR 0052). Importing it runs its top-level code and awaits any top-level await, so this
|
||||
* returns only once the step's own code has finished; a step that throws rejects here and the
|
||||
* process exits non-zero, which is how the host knows the step did not complete and must not start
|
||||
* the container it gates. No broker is connected — a first-boot seed or migration runs offline.
|
||||
*/
|
||||
async function runEntry(entrypoint: string): Promise<void> {
|
||||
if (!entrypoint) {
|
||||
console.error("mesh-tools run <entrypoint> — a compiled module entrypoint path is required");
|
||||
process.exit(1);
|
||||
}
|
||||
// A file URL, not a bare path: dynamic import of an absolute path is not portable, and the
|
||||
// entrypoint the manifest names is an absolute path inside the image.
|
||||
await import(pathToFileURL(entrypoint).href);
|
||||
}
|
||||
|
||||
/**
|
||||
* Bring this module's state to the shape this version needs, and exit — the runtime's answer to the
|
||||
* one word the mesh asks every module (novox/hq ADR 0135).
|
||||
*
|
||||
* The entrypoints come from `MESH_PREPARE`, which a module's own image sets beside the entrypoints it
|
||||
* already lists there: the module knows which of its files prepares its state, and nothing else could.
|
||||
* Each is imported in the order given, to completion, with no broker — preparation runs before the
|
||||
* version that would use it, so there is nothing yet to talk to.
|
||||
*
|
||||
* **An empty list is a failure, not a no-op.** The mesh only asks this of a module whose manifest says
|
||||
* it prepares something; a module that says so and names nothing has been built wrong, and exiting 0
|
||||
* would let that version serve against a state nobody shaped.
|
||||
*/
|
||||
async function prepareState(): Promise<void> {
|
||||
const named = (process.env.MESH_PREPARE ?? "")
|
||||
.split(",")
|
||||
.map((entry) => entry.trim())
|
||||
.filter((entry) => entry !== "");
|
||||
if (named.length === 0) {
|
||||
console.error(
|
||||
"mesh-tools prepare: this module was asked to prepare its state and its image names nothing " +
|
||||
"to do it with — set MESH_PREPARE to the compiled entrypoint(s) that prepare it, the way " +
|
||||
"MESH_TOOL_MODULES names the ones it serves",
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
for (const entrypoint of named) {
|
||||
console.log(`[mesh-tools] preparing with ${entrypoint}`);
|
||||
await runEntry(entrypoint);
|
||||
}
|
||||
console.log(`[mesh-tools] prepared: ${named.length} entrypoint(s) ran to completion`);
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const [command, ...rest] = process.argv.slice(2);
|
||||
if (command === "run") {
|
||||
await runEntry(rest[0] ?? "");
|
||||
return;
|
||||
}
|
||||
if (command === "prepare") {
|
||||
await prepareState();
|
||||
return;
|
||||
}
|
||||
if (command === "invoke") {
|
||||
const [module, tool] = rest;
|
||||
if (!module || !tool) {
|
||||
console.error("mesh-tools invoke <module> <tool> [json-args] — a module and tool are required");
|
||||
process.exit(1);
|
||||
}
|
||||
await invokeOnce(module, tool, rest[2] ?? "");
|
||||
return;
|
||||
}
|
||||
if (command === "emit") {
|
||||
const type = rest[0];
|
||||
if (!type) {
|
||||
console.error("mesh-tools emit <type> [json-body] — a routing key is required");
|
||||
process.exit(1);
|
||||
}
|
||||
await emitOnce(type, rest[1] ?? "");
|
||||
return;
|
||||
}
|
||||
await serve();
|
||||
}
|
||||
|
||||
// Only when run, so a test can import the pieces.
|
||||
if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
|
||||
void main();
|
||||
}
|
||||
@@ -0,0 +1,258 @@
|
||||
/**
|
||||
* The mesh's tools as an MCP server (novox/hq design 25 §7, design 34).
|
||||
*
|
||||
* **A thin adapter and nothing more.** Every tool an agent sees is one a module answered for and one
|
||||
* this account may call; the schema is the module's own; the answer is the module's own. Nothing here
|
||||
* decides anything, which is why it is short — an MCP surface that reshaped arguments or summarised
|
||||
* answers would be a second definition of what a tool is, and the module's code is the first.
|
||||
*
|
||||
* Implemented against the protocol directly rather than through a library: the surface is three
|
||||
* methods and one framing, and a dependency here would be a dependency on every machine.
|
||||
*
|
||||
* One handler, two transports. Over stdio for a program a person starts (`mesh mcp`), over HTTP on a
|
||||
* machine's loopback for the console the mesh assigns there (`mesh serve`, http.ts). The handler does
|
||||
* not know which asked.
|
||||
*/
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
import { callTool, seatsIn, toolKey, toolsOn, whyItFailed, type Listing, type Seats } from "./client.js";
|
||||
|
||||
/** The protocol version this speaks. Stated, because a host that wants another should be told so
|
||||
* rather than discovering it through a shape it did not expect. */
|
||||
export const PROTOCOL = "2025-03-26";
|
||||
|
||||
export interface Request {
|
||||
jsonrpc: string;
|
||||
id?: number | string | null;
|
||||
method: string;
|
||||
params?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface Reply {
|
||||
jsonrpc: "2.0";
|
||||
id: Request["id"];
|
||||
result?: unknown;
|
||||
error?: { code: number; message: string };
|
||||
}
|
||||
|
||||
/** How long a fetched tool list is kept before the modules are asked again. An agent asks on every
|
||||
* turn; the mesh changes on the order of minutes. */
|
||||
export const LISTING_KEPT_MS = 30_000;
|
||||
|
||||
export interface Surface {
|
||||
/** Answer one request; undefined for a notification, which expects none. */
|
||||
handle(request: Request): Promise<Reply | undefined>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The surface over one bus connection, as one account.
|
||||
*
|
||||
* The tool list is fetched when first asked and kept for a short while (design 34 §3): asking every
|
||||
* module on every `tools/list` would fan out on every agent turn for something nobody changed, and
|
||||
* never refreshing would hide a module assigned a moment ago.
|
||||
*/
|
||||
export function mcpSurface(bus: Broker, who: string): Surface {
|
||||
let known: { listing: Listing; at: number } | undefined;
|
||||
|
||||
const answer = (id: Request["id"], result: unknown): Reply => ({ jsonrpc: "2.0", id, result });
|
||||
const refuse = (id: Request["id"], code: number, message: string): Reply => ({
|
||||
jsonrpc: "2.0",
|
||||
id,
|
||||
error: { code, message },
|
||||
});
|
||||
|
||||
const listing = async (): Promise<Listing> => {
|
||||
if (!known || Date.now() - known.at > LISTING_KEPT_MS) {
|
||||
known = { listing: await toolsOn(bus), at: Date.now() };
|
||||
}
|
||||
return known.listing;
|
||||
};
|
||||
// The roles the last listing knew, so `<seat>.<verb>` resolves to the seat. Fetched once if a
|
||||
// call arrives before any list did; a listing that failed leaves no roles, and the name is then
|
||||
// a module's, which is the right fallback for a mesh whose control plane is away.
|
||||
const roles = async (): Promise<Seats | undefined> => {
|
||||
try {
|
||||
return seatsIn(await listing());
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
async handle(request) {
|
||||
// A notification has no id and expects no answer; `initialized` is the one every host sends.
|
||||
const notification = request.id === undefined || request.id === null;
|
||||
|
||||
switch (request.method) {
|
||||
case "initialize":
|
||||
return answer(request.id, {
|
||||
protocolVersion: PROTOCOL,
|
||||
capabilities: { tools: {} },
|
||||
serverInfo: { name: "mesh", version: "1" },
|
||||
// Said in the handshake, because an agent that knows whose authority it is acting under
|
||||
// can say so when a call is refused — and a refusal is the one thing here that is not
|
||||
// the mesh's fault or the tool's.
|
||||
instructions:
|
||||
`These are the tools of a Novox mesh, reached as ${who}. Every call goes to the module ` +
|
||||
`that serves it; what may be called was fixed when this account was issued, so a ` +
|
||||
`refusal means the account, not the tool. The list is what the running modules ` +
|
||||
`answered, plus every role's tools from the mesh's records — the mesh's own verbs ` +
|
||||
`(mesh-controller.status, .push, .assign …) among them; a module that did not answer ` +
|
||||
`is named in the list's _meta and can still be called by <module>.<tool>.`,
|
||||
});
|
||||
|
||||
case "notifications/initialized":
|
||||
return undefined;
|
||||
|
||||
case "ping":
|
||||
return notification ? undefined : answer(request.id, {});
|
||||
|
||||
case "tools/list": {
|
||||
let have: Listing;
|
||||
try {
|
||||
have = await listing();
|
||||
} catch (e) {
|
||||
return refuse(request.id, -32603, whyItFailed("mesh-catalog.catalog_modules", e));
|
||||
}
|
||||
return answer(request.id, {
|
||||
tools: have.tools.map((t) => ({
|
||||
name: `${t.module}.${t.name}`,
|
||||
description: t.description ?? `${t.name}, served by ${t.module}`,
|
||||
// The module's own schema, passed through — with `node`, the machine to ask when
|
||||
// the module runs on several (novox/hq ADR 0159); a seat's verb takes none, the
|
||||
// seat's scope decides. An empty object is a tool that takes nothing, which is a
|
||||
// real answer and not a missing one.
|
||||
inputSchema: t.seat && t.scope !== "node" ? asSchema(t.input)
|
||||
: t.seat ? withNode(asSchema(t.input), "the machine whose seat answers; required, the seat is held once per machine", true)
|
||||
: withNode(asSchema(t.input)),
|
||||
})),
|
||||
// Silence, named (design 34 §3): the modules the catalogue holds and nothing answered
|
||||
// for. Not a tool, so not in `tools`; not dropped either.
|
||||
_meta: { notAnswering: have.notAnswering },
|
||||
});
|
||||
}
|
||||
|
||||
case "tools/call": {
|
||||
const given = String(request.params?.name ?? "");
|
||||
const args = { ...((request.params?.arguments as Record<string, unknown> | undefined) ?? {}) };
|
||||
// The machine, when the caller names one, travels in the subject and never reaches the
|
||||
// module's arguments (novox/hq ADR 0159) — for a module's tool. A seat's verb takes no
|
||||
// machine from the console (the seat's scope decides), so a `node` among its arguments
|
||||
// is the verb's own, as `push` and `assign` take one, and is handed through untouched.
|
||||
const have = await listing().catch(() => undefined);
|
||||
const roles = have && seatsIn(have);
|
||||
const bare = given.split("@", 1)[0];
|
||||
const isSeatVerb = roles ? toolKey(bare, roles).startsWith("seat:") : false;
|
||||
// A node-scoped seat's verb is asked of one machine (design 33 §4, ADR 0170): `node`
|
||||
// names it and travels in the subject, as for a module's tool.
|
||||
const nodeScoped = isSeatVerb && (have?.tools.some((t) => t.seat && t.scope === "node" &&
|
||||
`${t.module}.${t.name}` === bare) ?? false);
|
||||
const takesNode = !isSeatVerb || nodeScoped;
|
||||
const node = takesNode && typeof args.node === "string" && args.node !== "" ? args.node : "";
|
||||
if (takesNode) delete args.node;
|
||||
if (nodeScoped && !node && !given.includes("@")) {
|
||||
return refuse(request.id, -32602, `${given} is a machine's seat's verb: name the machine with \`node\``);
|
||||
}
|
||||
const name = node && !given.includes("@") ? `${given}@${node}` : given;
|
||||
try {
|
||||
const { result, node: answeredBy } = await callTool(bus, name, args, roles, have);
|
||||
// Text, because that is what every host renders. The content is the module's answer
|
||||
// as JSON, unshaped: an adapter that flattened it would be deciding what matters in
|
||||
// somebody else's answer. Which machine answered follows it as its own line.
|
||||
const content: { type: string; text: string }[] = [
|
||||
{ type: "text", text: JSON.stringify(result, null, 2) },
|
||||
];
|
||||
if (answeredBy) content.push({ type: "text", text: `answered by ${answeredBy}` });
|
||||
return answer(request.id, { content });
|
||||
} catch (e) {
|
||||
// **An error the agent can act on, not a stack.** isError rather than a protocol
|
||||
// failure, because the call was well-formed and the mesh answered it — with a refusal,
|
||||
// an absence or a fault, and the words say which.
|
||||
return answer(request.id, {
|
||||
content: [{ type: "text", text: whyItFailed(name, e) }],
|
||||
isError: true,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
default:
|
||||
return notification ? undefined : refuse(request.id, -32601, `mesh's MCP surface has no ${request.method}`);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* A module's declared input as a JSON schema an agent can read.
|
||||
*
|
||||
* The sdk keeps a tool's input opaque, and the catalogue's modules write it as a bare map of
|
||||
* property to description — `{ module: { type, description } }` — which is the `properties` of a
|
||||
* schema rather than a schema. Wrapped here when that is what arrived; passed through when a module
|
||||
* already wrote a schema; an empty object when it declared nothing. The module's words are kept
|
||||
* either way.
|
||||
*/
|
||||
export function asSchema(input: unknown): Record<string, unknown> {
|
||||
if (!input || typeof input !== "object" || Array.isArray(input)) {
|
||||
return { type: "object", properties: {} };
|
||||
}
|
||||
const given = input as Record<string, unknown>;
|
||||
if (given.type === "object" || "properties" in given) return given;
|
||||
if (Object.keys(given).length === 0) return { type: "object", properties: {} };
|
||||
return { type: "object", properties: given };
|
||||
}
|
||||
|
||||
/**
|
||||
* Serve over stdio until stdin closes, which is how a host ends a session.
|
||||
*/
|
||||
export async function serveMcp(bus: Broker, who: string): Promise<void> {
|
||||
const surface = mcpSurface(bus, who);
|
||||
const say = (message: unknown) => {
|
||||
process.stdout.write(`${JSON.stringify(message)}\n`);
|
||||
};
|
||||
for await (const line of lines()) {
|
||||
let request: Request;
|
||||
try {
|
||||
request = JSON.parse(line) as Request;
|
||||
} catch {
|
||||
// Unparseable, and with no id there is nobody to tell. Skipped rather than answered, because a
|
||||
// reply to a request that was never framed is noise on the same channel.
|
||||
continue;
|
||||
}
|
||||
const reply = await surface.handle(request);
|
||||
if (reply) say(reply);
|
||||
}
|
||||
}
|
||||
|
||||
/** stdin as newline-framed messages, which is what MCP over stdio is. */
|
||||
async function* lines(): AsyncGenerator<string> {
|
||||
let buffered = "";
|
||||
for await (const chunk of process.stdin) {
|
||||
buffered += (chunk as Buffer).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 !== "") yield line;
|
||||
}
|
||||
}
|
||||
if (buffered.trim() !== "") yield buffered.trim();
|
||||
}
|
||||
|
||||
/** Every module tool takes an optional `node`: the machine to ask when the module runs on several
|
||||
* (novox/hq ADR 0159). Added to the listing, stripped before the call, never seen by the module. */
|
||||
function withNode(schema: Record<string, unknown>, description?: string, required = false): Record<string, unknown> {
|
||||
const properties = { ...((schema.properties as Record<string, unknown> | undefined) ?? {}) };
|
||||
if (!("node" in properties)) {
|
||||
properties.node = {
|
||||
type: "string",
|
||||
description: description ??
|
||||
"the machine to ask, when this module runs on several; else whichever answers, and the answer says which",
|
||||
};
|
||||
}
|
||||
const out: Record<string, unknown> = { ...schema, type: "object", properties };
|
||||
if (required) {
|
||||
const have = Array.isArray(schema.required) ? (schema.required as string[]) : [];
|
||||
out.required = have.includes("node") ? have : [...have, "node"];
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* `mesh` — the mesh's tools, for whoever is on a machine (novox/hq design 25 §7, design 34).
|
||||
*
|
||||
* Four verbs and nothing else. What tools are there, call one, serve the same two to an agent over
|
||||
* stdio, and serve them on a machine's loopback as the console the mesh assigns. Deliberately thin:
|
||||
* everything that could be a decision is one the mesh already made, and a client that grew opinions
|
||||
* would be a second place the mesh's behaviour is defined.
|
||||
*
|
||||
* mesh tools what the running modules answer
|
||||
* mesh call <module>.<tool> [json] call one, arguments as JSON on the command line or on stdin
|
||||
* mesh mcp the same two, as an MCP server over stdio
|
||||
* mesh serve [--listen host:port] the console: MCP over HTTP on this machine's loopback
|
||||
*
|
||||
* Who it speaks as, in order of preference:
|
||||
* MESH_BROKER_FILE the module credential the mesh delivered — the console's (ADR 0152)
|
||||
* MESH_CREDENTIAL / --credential <file> a person's, as `operator issue` printed it (design 25 §7)
|
||||
* MESH_CONSOLE / --console <url> no credential: `tools` and `call` go through a console
|
||||
* already running on this machine, over loopback HTTP
|
||||
*/
|
||||
import { readFile } from "node:fs/promises";
|
||||
|
||||
import type { Broker } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
import {
|
||||
callTool,
|
||||
connectAs,
|
||||
connectAsTheConsole,
|
||||
credentialFrom,
|
||||
seatsIn,
|
||||
toolsOn,
|
||||
whyItFailed,
|
||||
type Listing,
|
||||
} from "./client.js";
|
||||
import { serveMcpHttp } from "./http.js";
|
||||
import { serveMcp } from "./mcp.js";
|
||||
|
||||
const usage = `mesh tools
|
||||
mesh call <module>.<tool> [json]
|
||||
mesh mcp
|
||||
mesh serve [--listen host:port]
|
||||
|
||||
--credential <file> a person's credential, the JSON \`operator issue\` printed; default $MESH_CREDENTIAL
|
||||
--console <url> a console on this machine to ask through instead; default $MESH_CONSOLE
|
||||
--listen <host:port> where \`serve\` listens; loopback only; default $MESH_CONSOLE_LISTEN or 127.0.0.1:4270
|
||||
MESH_BROKER_FILE the module credential the mesh delivered, which \`serve\` runs on`;
|
||||
|
||||
/** What the console listens on when nothing says otherwise. */
|
||||
const DEFAULT_LISTEN = "127.0.0.1:4270";
|
||||
|
||||
async function main(argv: string[]): Promise<number> {
|
||||
const args = [...argv];
|
||||
let credentialPath = process.env.MESH_CREDENTIAL ?? "";
|
||||
let consoleUrl = process.env.MESH_CONSOLE ?? "";
|
||||
let listen = process.env.MESH_CONSOLE_LISTEN ?? DEFAULT_LISTEN;
|
||||
for (let i = 0; i < args.length; i++) {
|
||||
const take = () => {
|
||||
const v = args[i + 1] ?? "";
|
||||
args.splice(i, 2);
|
||||
i--;
|
||||
return v;
|
||||
};
|
||||
if (args[i] === "--credential") credentialPath = take();
|
||||
else if (args[i] === "--console") consoleUrl = take();
|
||||
else if (args[i] === "--listen") listen = take();
|
||||
}
|
||||
const verb = args.shift();
|
||||
if (!verb || verb === "help" || verb === "--help") {
|
||||
console.log(usage);
|
||||
return verb ? 0 : 1;
|
||||
}
|
||||
|
||||
// Through a console already on this machine: no credential to hold, which is the point of one.
|
||||
if (consoleUrl && (verb === "tools" || verb === "call")) {
|
||||
return verb === "tools" ? listingVia(consoleUrl) : callingVia(consoleUrl, args);
|
||||
}
|
||||
|
||||
const { bus, who } = await connecting(credentialPath, verb);
|
||||
try {
|
||||
switch (verb) {
|
||||
case "tools":
|
||||
return await listing(bus, who);
|
||||
case "call":
|
||||
return await calling(bus, args);
|
||||
case "mcp":
|
||||
// Serves until stdin closes, which is how an MCP host ends a session.
|
||||
await serveMcp(bus, who);
|
||||
return 0;
|
||||
case "serve": {
|
||||
const up = await serveMcpHttp(bus, who, listen);
|
||||
console.log(`mesh console listening on http://${up.address}/mcp as ${who}`);
|
||||
await new Promise<void>((resolve) => {
|
||||
process.once("SIGTERM", () => resolve());
|
||||
process.once("SIGINT", () => resolve());
|
||||
});
|
||||
await up.close();
|
||||
return 0;
|
||||
}
|
||||
default:
|
||||
console.error(`mesh has no "${verb}".\n\n${usage}`);
|
||||
return 1;
|
||||
}
|
||||
} finally {
|
||||
await bus.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Who this process is on the bus. The module credential first: a console is started by the mesh with
|
||||
* MESH_BROKER_FILE and nothing else, and must not fall back to a person's file lying around.
|
||||
*/
|
||||
async function connecting(credentialPath: string, verb: string): Promise<{ bus: Broker; who: string }> {
|
||||
const delivered = process.env.MESH_BROKER_FILE;
|
||||
if (delivered) {
|
||||
return connectAsTheConsole(delivered);
|
||||
}
|
||||
if (!credentialPath) {
|
||||
throw new Error(
|
||||
verb === "serve"
|
||||
? "no credential: the console runs on MESH_BROKER_FILE, the module credential the mesh " +
|
||||
"delivered; to run it by hand, pass --credential <file> with a person's credential"
|
||||
: "no credential: set MESH_CREDENTIAL or pass --credential <file> (the JSON `operator issue` " +
|
||||
"printed, saved verbatim), or --console <url> to ask through a console on this machine",
|
||||
);
|
||||
}
|
||||
const held = await credentialFrom(credentialPath);
|
||||
return { bus: await connectAs(held), who: held.person ?? held.user ?? "somebody" };
|
||||
}
|
||||
|
||||
function printListing(have: Listing, who?: string): void {
|
||||
if (have.tools.length === 0) {
|
||||
console.log("no running module answered with any tool");
|
||||
}
|
||||
// **What the modules answered, not what this account may call.** The two differ and the
|
||||
// difference is the point: an account seeing only its own tools cannot tell "not installed" from
|
||||
// "not yours", and those need different people to fix them.
|
||||
for (const t of have.tools) {
|
||||
const name = `${t.module}.${t.name}`;
|
||||
const line = t.description ? `${name.padEnd(36)} ${t.description}` : name;
|
||||
console.log(t.seat ? `${line} (a role's tool: answered by whoever holds the ${t.module} seat)` : line);
|
||||
}
|
||||
if (have.notAnswering.length > 0) {
|
||||
console.log(
|
||||
`\nheld by the mesh and not answering: ${have.notAnswering.join(", ")} — not assigned, not up, ` +
|
||||
"or built before the runtime answered `tools`; each can still be called by name",
|
||||
);
|
||||
}
|
||||
if (who) {
|
||||
console.log(`\nasked as ${who}; what ${who} may call was fixed when the account was issued.`);
|
||||
}
|
||||
}
|
||||
|
||||
async function listing(bus: Broker, who?: string): Promise<number> {
|
||||
let have: Listing;
|
||||
try {
|
||||
have = await toolsOn(bus);
|
||||
} catch (e) {
|
||||
console.error(whyItFailed("mesh-catalog.catalog_modules", e));
|
||||
return 1;
|
||||
}
|
||||
printListing(have, who);
|
||||
return 0;
|
||||
}
|
||||
|
||||
async function calling(bus: Broker, args: string[]): Promise<number> {
|
||||
const key = args.shift();
|
||||
if (!key) {
|
||||
console.error("mesh call <module>.<tool> [json]");
|
||||
return 1;
|
||||
}
|
||||
const parsed = await argumentsFrom(args);
|
||||
if (parsed === undefined) return 1;
|
||||
try {
|
||||
// `<seat>.<verb>` reaches the role when the mesh lists that verb for the seat; `seat:` says so
|
||||
// outright and asks nothing first.
|
||||
const have = key.startsWith("seat:") ? undefined : await toolsOn(bus).catch(() => undefined);
|
||||
const { result, node } = await callTool(bus, key, parsed, have && seatsIn(have), have);
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
if (node) console.error(`answered by ${node}`);
|
||||
return 0;
|
||||
} catch (e) {
|
||||
console.error(whyItFailed(key, e));
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
/** The console's answer to one MCP request, over loopback HTTP. */
|
||||
async function viaConsole(consoleUrl: string, method: string, params?: unknown): Promise<any> {
|
||||
const endpoint = consoleUrl.endsWith("/mcp") ? consoleUrl : `${consoleUrl.replace(/\/$/, "")}/mcp`;
|
||||
const res = await fetch(endpoint, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json", accept: "application/json" },
|
||||
body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params }),
|
||||
});
|
||||
if (!res.ok) {
|
||||
throw new Error(`the console at ${endpoint} answered ${res.status}`);
|
||||
}
|
||||
const reply = (await res.json()) as { result?: any; error?: { message: string } };
|
||||
if (reply.error) throw new Error(reply.error.message);
|
||||
return reply.result;
|
||||
}
|
||||
|
||||
async function listingVia(consoleUrl: string): Promise<number> {
|
||||
try {
|
||||
const result = await viaConsole(consoleUrl, "tools/list");
|
||||
const have: Listing = {
|
||||
tools: (result.tools ?? []).map((t: { name: string; description?: string; inputSchema?: unknown }) => {
|
||||
const at = t.name.indexOf(".");
|
||||
return { module: t.name.slice(0, at), name: t.name.slice(at + 1), description: t.description, input: t.inputSchema };
|
||||
}),
|
||||
notAnswering: result._meta?.notAnswering ?? [],
|
||||
};
|
||||
printListing(have);
|
||||
return 0;
|
||||
} catch (e) {
|
||||
console.error(e instanceof Error ? e.message : String(e));
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
async function callingVia(consoleUrl: string, args: string[]): Promise<number> {
|
||||
const key = args.shift();
|
||||
if (!key) {
|
||||
console.error("mesh call <module>.<tool> [json]");
|
||||
return 1;
|
||||
}
|
||||
const parsed = await argumentsFrom(args);
|
||||
if (parsed === undefined) return 1;
|
||||
try {
|
||||
const result = await viaConsole(consoleUrl, "tools/call", { name: key, arguments: parsed });
|
||||
const text = result?.content?.[0]?.text ?? JSON.stringify(result);
|
||||
if (result?.isError) {
|
||||
console.error(text);
|
||||
return 1;
|
||||
}
|
||||
console.log(text);
|
||||
return 0;
|
||||
} catch (e) {
|
||||
console.error(e instanceof Error ? e.message : String(e));
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
/** A call's arguments: JSON on the command line, else on stdin, else nothing. Undefined when what
|
||||
* was given is not JSON, after saying so. */
|
||||
async function argumentsFrom(args: string[]): Promise<unknown> {
|
||||
const raw = args.length > 0 ? args.join(" ") : await maybeStdin();
|
||||
if (raw.trim() === "") return {};
|
||||
try {
|
||||
return JSON.parse(raw);
|
||||
} catch (e) {
|
||||
console.error(`the arguments are not JSON: ${(e as Error).message}`);
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** Arguments on stdin, for a call whose JSON is too long or too quoted to type. Empty when stdin is a
|
||||
* terminal, so `mesh call x.y` with no arguments does not hang waiting for something nobody is
|
||||
* typing. */
|
||||
async function maybeStdin(): Promise<string> {
|
||||
if (process.stdin.isTTY) return "";
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of process.stdin) chunks.push(chunk as Buffer);
|
||||
return Buffer.concat(chunks).toString("utf8");
|
||||
}
|
||||
|
||||
// Only when run, so a test can import the pieces.
|
||||
if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
|
||||
main(process.argv.slice(2))
|
||||
.then((code) => process.exit(code))
|
||||
.catch((e) => {
|
||||
console.error(e instanceof Error ? e.message : String(e));
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
|
||||
export { main, usage };
|
||||
export const _readFile = readFile;
|
||||
@@ -0,0 +1,315 @@
|
||||
// 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());
|
||||
}
|
||||
Reference in New Issue
Block a user