303 lines
14 KiB
TypeScript
303 lines
14 KiB
TypeScript
/**
|
|
* A tools bundle as a process the node's runtime launches (novox/hq ADR 0188).
|
|
*
|
|
* The runtime speaks MCP over stdio to it: `initialize`, `tools/list` once, `tools/call` per call,
|
|
* newline-framed JSON-RPC on stdin and stdout. This is the whole of what a tools bundle in any
|
|
* language must do, and this file is the TypeScript SDK's share of it: the loop, so a module writes
|
|
* its tools and nothing else. A tool named `<seat>.<verb>` is the module's implementation of that
|
|
* seat's verb; any other name is the module's own tool.
|
|
*
|
|
* **No transport here.** The bus, the subjects, the memberships are the runtime's (ADR 0039, 0175);
|
|
* a bundle knows only stdin and stdout, which is why a bus change rebuilds no bundle in any language.
|
|
*
|
|
* stdout is the protocol. A tool that wants to log writes to stderr, which the runtime keeps.
|
|
*/
|
|
import type { ToolDefinition } from "../contracts/index.js";
|
|
import { collectTools } from "../tools/index.js";
|
|
import { useBroker, type Broker, type Envelope } from "../messaging/index.js";
|
|
import { useStateChannel, type StateChange } from "../state/index.js";
|
|
|
|
/** The protocol version this speaks; the runtime says the same. */
|
|
export const PROTOCOL = "2025-03-26";
|
|
|
|
interface Request {
|
|
jsonrpc: string;
|
|
id?: number | string | null;
|
|
method: string;
|
|
params?: Record<string, unknown>;
|
|
}
|
|
|
|
/** A module's declared input as a JSON schema: the bare property map the catalogue's modules write
|
|
* is wrapped, a schema is passed through, nothing is an empty object. */
|
|
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 };
|
|
}
|
|
|
|
/**
|
|
* One stdio session per process: the channel to the node's runtime that launched this bundle
|
|
* (novox/hq ADR 0193, ADR 0198). It answers MCP — initialize, tools/list, tools/call — and is this
|
|
* bundle's bus: what the module's code emits, subscribes to and asks goes to the runtime as a request
|
|
* on the same channel, and the runtime's answers and the events it delivers arrive on it.
|
|
*
|
|
* **Started before the bundle's own code runs.** A long-running entrypoint subscribes as it is
|
|
* imported (`await on(...)` at the top level), and the launcher imports it before it says what tools
|
|
* it serves — so the broker must already be bound and stdin already read when that import starts, or
|
|
* the subscription would wait on an answer nobody reads. The session therefore starts as this module
|
|
* loads when the runtime launched the process, and `initialize` and `tools/list` are answered once the
|
|
* bundle has said what it serves.
|
|
*/
|
|
interface Session {
|
|
serve(name: string, tools: ToolDefinition[]): void;
|
|
closed: Promise<void>;
|
|
}
|
|
|
|
let session: Session | undefined;
|
|
|
|
function startSession(): Session {
|
|
if (session) return session;
|
|
const say = (message: unknown): void => {
|
|
process.stdout.write(`${JSON.stringify(message)}\n`);
|
|
};
|
|
let nextOut = 1;
|
|
const awaiting = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
|
|
const askAs = (id: string, method: string, params: unknown): Promise<unknown> =>
|
|
new Promise((resolve, reject) => {
|
|
awaiting.set(id, { resolve, reject });
|
|
say({ jsonrpc: "2.0", id, method, params });
|
|
});
|
|
const ask = (method: string, params: unknown): Promise<unknown> => askAs(`bundle-${nextOut++}`, method, params);
|
|
|
|
// The module's state (ADR 0201): each watch's changes arrive as `mesh/state` requests naming the
|
|
// watch by the id it was asked with, so its handler is registered under that id before it is asked
|
|
// — the current values arrive before the watch itself is answered.
|
|
const watches = new Map<string, (change: StateChange<unknown>) => Promise<void>>();
|
|
useStateChannel({
|
|
ask,
|
|
watch: async (params, deliver) => {
|
|
const id = `bundle-${nextOut++}`;
|
|
watches.set(id, deliver);
|
|
try {
|
|
await askAs(id, "mesh/state.watch", params);
|
|
} catch (err) {
|
|
watches.delete(id);
|
|
throw err;
|
|
}
|
|
},
|
|
});
|
|
|
|
// What the module's code reacts to: every handler, matched here against each event the runtime
|
|
// delivers. The runtime binds the module's one durable consumer and hands every message it reads to
|
|
// this process; an event no handler is for is answered at once, so it is acknowledged rather than
|
|
// redelivered until it expires — the consumer's filter is the controller's and may be wider.
|
|
const listeners: { pattern: string; handler: (env: Envelope<unknown>) => Promise<void> }[] = [];
|
|
const viaRuntime: Broker = {
|
|
publish: async <T>(env: Envelope<T>) => {
|
|
await ask("mesh/publish", env);
|
|
},
|
|
request: (async <Req, Res>(key: string, body: Req): Promise<Res> =>
|
|
(await ask("mesh/ask", { key, body })) as Res) as Broker["request"],
|
|
handle: (async () => {
|
|
throw new Error("a launched bundle answers through its tools, not by binding subjects (novox/hq ADR 0193)");
|
|
}) as Broker["handle"],
|
|
subscribe: (async <T>(pattern: string, handler: (env: Envelope<T>) => Promise<void>) => {
|
|
const listener = { pattern, handler: handler as (env: Envelope<unknown>) => Promise<void> };
|
|
listeners.push(listener);
|
|
await ask("mesh/subscribe", { pattern });
|
|
return () => {
|
|
const at = listeners.indexOf(listener);
|
|
if (at >= 0) listeners.splice(at, 1);
|
|
};
|
|
}) as Broker["subscribe"],
|
|
close: async () => {},
|
|
};
|
|
useBroker(() => viaRuntime);
|
|
|
|
let name = "bundle";
|
|
let byName = new Map<string, ToolDefinition>();
|
|
let tools: ToolDefinition[] = [];
|
|
let ready!: () => void;
|
|
const served = new Promise<void>((resolve) => (ready = resolve));
|
|
|
|
const answer = async (request: Request): Promise<void> => {
|
|
const notification = request.id === undefined || request.id === null;
|
|
switch (request.method) {
|
|
case "initialize":
|
|
await served;
|
|
say({ jsonrpc: "2.0", id: request.id, result: { protocolVersion: PROTOCOL, capabilities: { tools: {} }, serverInfo: { name, version: "1" } } });
|
|
return;
|
|
case "notifications/initialized":
|
|
return;
|
|
case "ping":
|
|
if (!notification) say({ jsonrpc: "2.0", id: request.id, result: {} });
|
|
return;
|
|
case "tools/list":
|
|
await served;
|
|
say({
|
|
jsonrpc: "2.0", id: request.id,
|
|
result: { tools: tools.map((t) => ({ name: t.name, description: t.description, inputSchema: asSchema(t.input) })) },
|
|
});
|
|
return;
|
|
case "tools/call": {
|
|
await served;
|
|
const wanted = String(request.params?.name ?? "");
|
|
const tool = byName.get(wanted);
|
|
if (!tool) {
|
|
say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: `${name} has no tool ${wanted}` } });
|
|
return;
|
|
}
|
|
try {
|
|
const result = await tool.run((request.params?.arguments as Record<string, unknown> | undefined) ?? {});
|
|
say({ jsonrpc: "2.0", id: request.id, result: { content: [{ type: "text", text: JSON.stringify(result ?? null) }] } });
|
|
} catch (err) {
|
|
// The tool failed; said as a tool error, not a protocol one, because the call was well-formed.
|
|
say({ jsonrpc: "2.0", id: request.id, result: { content: [{ type: "text", text: err instanceof Error ? err.message : String(err) }], isError: true } });
|
|
}
|
|
return;
|
|
}
|
|
case "mesh/event": {
|
|
// One event the runtime read from this module's consumer (ADR 0198). Answered once every
|
|
// handler it is for has taken it; an error is answered as one, so the runtime does not
|
|
// acknowledge it and the bus delivers it again.
|
|
const env = (request.params as { envelope?: Envelope<unknown> } | undefined)?.envelope;
|
|
if (!env || typeof env.key !== "string") {
|
|
say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/event carries no envelope" } });
|
|
return;
|
|
}
|
|
try {
|
|
for (const l of listeners.filter((x) => topicMatches(x.pattern, env.key))) await l.handler(env);
|
|
say({ jsonrpc: "2.0", id: request.id, result: {} });
|
|
} catch (err) {
|
|
say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
|
|
}
|
|
return;
|
|
}
|
|
case "mesh/state": {
|
|
// One change to a state this bundle watches (ADR 0201), answered once its handler has taken
|
|
// it. The runtime delivers a watch's changes one at a time, in order.
|
|
const p = request.params as { watch?: string; change?: StateChange<unknown> } | undefined;
|
|
const deliver = p?.watch !== undefined ? watches.get(String(p.watch)) : undefined;
|
|
if (!deliver || !p?.change) {
|
|
say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/state names no watch of this bundle" } });
|
|
return;
|
|
}
|
|
try {
|
|
await deliver(p.change);
|
|
say({ jsonrpc: "2.0", id: request.id, result: {} });
|
|
} catch (err) {
|
|
say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
|
|
}
|
|
return;
|
|
}
|
|
default:
|
|
if (!notification) say({ jsonrpc: "2.0", id: request.id, error: { code: -32601, message: `no ${request.method}` } });
|
|
}
|
|
};
|
|
|
|
const closed = (async () => {
|
|
for await (const line of lines()) {
|
|
let request: Request;
|
|
try {
|
|
request = JSON.parse(line) as Request;
|
|
} catch {
|
|
continue; // not framed; nobody to answer
|
|
}
|
|
// The runtime's answer to something this bundle asked.
|
|
const reply = request as unknown as { id?: string; method?: string; result?: unknown; error?: { message?: string } };
|
|
if (reply.method === undefined && typeof reply.id === "string" && awaiting.has(reply.id)) {
|
|
const waiting = awaiting.get(reply.id)!;
|
|
awaiting.delete(reply.id);
|
|
if (reply.error) waiting.reject(new Error(reply.error.message ?? "the runtime refused"));
|
|
else waiting.resolve(reply.result);
|
|
continue;
|
|
}
|
|
// Beside the loop, never in it: a handler or tool that emits or asks waits for the runtime's
|
|
// answer, which arrives on the line this loop reads next — and nothing queues behind a slow one.
|
|
void answer(request);
|
|
}
|
|
})();
|
|
|
|
session = {
|
|
serve(n, ts) {
|
|
name = n;
|
|
tools = ts;
|
|
byName = new Map(ts.map((t) => [t.name, t]));
|
|
ready();
|
|
},
|
|
closed,
|
|
};
|
|
return session;
|
|
}
|
|
|
|
/**
|
|
* Serve these tools to the runtime over stdio until stdin closes. `name` is the module's; a tool
|
|
* whose name is `<seat>.<verb>` is served as the seat's. Resolves when the runtime hangs up. A bundle
|
|
* that only runs long serves no tools and still answers the runtime's handshake.
|
|
*/
|
|
export async function serveStdio(name: string, tools: ToolDefinition[]): Promise<void> {
|
|
const s = startSession();
|
|
s.serve(name, tools);
|
|
await s.closed;
|
|
}
|
|
|
|
/** Whether an event's key matches a handler's pattern: `*` one segment, `#` or `**` any number —
|
|
* the same rule the runtime's own subscriptions use (novox/hq design 29 §1). */
|
|
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;
|
|
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);
|
|
}
|
|
|
|
/** The word the node's runtime sets for a bundle it launches: the module it serves the bundle as
|
|
* (novox/hq ADR 0193). */
|
|
export const SERVED_AS = "MESH_SERVED_MODULE";
|
|
|
|
/**
|
|
* Serve everything registered with `registerModuleTools` over stdio — the launcher every TypeScript
|
|
* bundle the runtime serves is started through (novox/hq ADR 0193). What the module the runtime
|
|
* serves this bundle as registered is listed by its own name; anything registered under another
|
|
* name — a seat this module implements — is listed `<seat>.<verb>`, whichever was registered first.
|
|
* Every contributor is given this process's environment, which is the bundle's own.
|
|
*/
|
|
export async function serveRegisteredOverStdio(as: string | undefined = process.env[SERVED_AS]): Promise<void> {
|
|
const groups = collectTools(process.env);
|
|
const name = as ?? groups[0]?.module ?? "bundle";
|
|
const tools: ToolDefinition[] = [];
|
|
for (const g of groups) {
|
|
for (const t of g.tools) tools.push(g.module === name ? t : { ...t, name: `${g.module}.${t.name}` });
|
|
}
|
|
await serveStdio(name, tools);
|
|
}
|
|
|
|
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();
|
|
}
|
|
|
|
// Launched by the node's runtime: the channel is this process's from its first line, so the module's
|
|
// code may subscribe, emit and ask while it is being imported (ADR 0198).
|
|
if (process.env[SERVED_AS]) startSession();
|