novox/mesh-sdk
8
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

A launched bundle subscribes and asks through its runtime, and its code may do so while being imported (hq ADR 0198)

Browse Source 
The stdio channel is the bundle's bus: subscribe asks the runtime (mesh/subscribe), which binds the
module's durable consumer and hands each event over as mesh/event — answered once every matching
handler has taken it, as an error when one failed so the runtime does not acknowledge it; request
asks the runtime (mesh/ask). The session starts as the stdio module loads when the runtime launched
the process, because a handler module subscribes at import, before the launcher says what it
serves; initialize and tools/list are answered once it has. topicMatches as the runtime's. 0.1.6.
This commit is contained in:
jochen
2026-10-03 22:23:16 +02:00
parent 16370de2be
commit 1f936dc4a6
4 changed files with 230 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/stdio/index.ts
+149 -48
View File
@@ -37,18 +37,30 @@ export function asSchema(input: unknown): Record<string, unknown> {
}
/**
* 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.
* 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.
*/
export async function serveStdio(name: string, tools: ToolDefinition[]): Promise<void> {
const byName = new Map(tools.map((t) => [t.name, t]));
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`);
};
// **What a tool emits goes through the runtime** (novox/hq ADR 0193). A launched bundle holds no
// bus connection; the runtime that launched it does, and publishes as this module. So the broker a
// tool's `emit` reaches is this channel: a `mesh/publish` request the runtime answers once the bus
// has accepted it. Asking, answering and subscribing are the runtime's, not a tool's.
let nextOut = 1;
const awaiting = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
const ask = (method: string, params: unknown): Promise<unknown> =>
@@ -57,76 +69,161 @@ export async function serveStdio(name: string, tools: ToolDefinition[]): Promise
awaiting.set(id, { resolve, reject });
say({ jsonrpc: "2.0", id, method, params });
});
const refused = (what: string) => async (): Promise<never> => {
throw new Error(`a launched bundle does not ${what} on the bus; its runtime does (novox/hq ADR 0193)`);
};
// 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: refused("ask") as Broker["request"],
handle: refused("answer") as Broker["handle"],
subscribe: refused("subscribe") as Broker["subscribe"],
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);
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;
}
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" } } });
break;
return;
case "notifications/initialized":
break;
return;
case "ping":
if (!notification) say({ jsonrpc: "2.0", id: request.id, result: {} });
break;
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) })) },
});
break;
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}` } });
break;
return;
}
// Run beside the loop, never in it: a tool that emits waits for the runtime's answer, which
// arrives on the line this loop reads next — and calls do not queue behind a slow one.
const id = request.id;
void (async () => {
try {
const result = await tool.run((request.params?.arguments as Record<string, unknown> | undefined) ?? {});
say({ jsonrpc: "2.0", 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, result: { content: [{ type: "text", text: err instanceof Error ? err.message : String(err) }], isError: true } });
}
})();
break;
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;
}
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
@@ -163,3 +260,7 @@ async function* lines(): AsyncGenerator<string> {
}
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();