diff --git a/package.json b/package.json index fea7f88..e92b3aa 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.5", + "version": "0.1.6", "description": "The stable spine a Novox Mesh module's own code builds against.", "type": "module", "exports": { diff --git a/src/stdio/index.ts b/src/stdio/index.ts index 4cf4b07..ef5abe3 100644 --- a/src/stdio/index.ts +++ b/src/stdio/index.ts @@ -37,18 +37,30 @@ export function asSchema(input: unknown): Record { } /** - * Serve these tools to the runtime over stdio until stdin closes. `name` is the module's; a tool - * whose name is `.` 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 { - const byName = new Map(tools.map((t) => [t.name, t])); +interface Session { + serve(name: string, tools: ToolDefinition[]): void; + closed: Promise; +} + +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 void; reject: (e: Error) => void }>(); const ask = (method: string, params: unknown): Promise => @@ -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 => { - 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) => Promise }[] = []; const viaRuntime: Broker = { publish: async (env: Envelope) => { 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 (key: string, body: Req): Promise => + (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 (pattern: string, handler: (env: Envelope) => Promise) => { + const listener = { pattern, handler: handler as (env: Envelope) => Promise }; + 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(); + let tools: ToolDefinition[] = []; + let ready!: () => void; + const served = new Promise((resolve) => (ready = resolve)); + + const answer = async (request: Request): Promise => { 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 | 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 | 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 } | 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 `.` 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 { + 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 { } 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(); diff --git a/test/fixtures/stdio-long-running.mjs b/test/fixtures/stdio-long-running.mjs new file mode 100644 index 0000000..08d77bf --- /dev/null +++ b/test/fixtures/stdio-long-running.mjs @@ -0,0 +1,15 @@ +// A launched bundle whose module code runs long (novox/hq ADR 0198): it subscribes as it is imported, +// before it says what it serves — exactly as a handler module's index does — and one tool asks the bus. +import { on } from "../../dist/events/index.js"; +import { broker } from "../../dist/messaging/index.js"; +import { registerModuleTools } from "../../dist/tools/index.js"; +import { serveRegisteredOverStdio } from "../../dist/stdio/index.js"; + +await on("alpha.happened", async (e) => { + process.stderr.write(`handled ${e.type} ${JSON.stringify(e.body)}\n`); + if (e.body && e.body.fail) throw new Error("not this one"); +}); +registerModuleTools("watcher", () => [ + { name: "relay", description: "asks beta", input: {}, run: async () => broker().request("beta.one", { x: 1 }) }, +]); +await serveRegisteredOverStdio(); diff --git a/test/stdio.test.ts b/test/stdio.test.ts index 306778e..b76a5ad 100644 --- a/test/stdio.test.ts +++ b/test/stdio.test.ts @@ -104,3 +104,68 @@ test("a launched tool's emit is asked of the runtime, and the call answers once child.kill(); } }); + +test("a launched bundle subscribes as it is imported, handles events the runtime hands it, and asks through it (ADR 0198)", async () => { + const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-long-running.mjs", import.meta.url))], + { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "watcher" } }); + const seen: any[] = []; + let buffered = ""; + let said = ""; + child.stderr.on("data", (d) => (said += d.toString())); + child.stdout.on("data", (d) => { + buffered += d.toString(); + let at: number; + while ((at = buffered.indexOf("\n")) >= 0) { + seen.push(JSON.parse(buffered.slice(0, at))); + buffered = buffered.slice(at + 1); + } + }); + const until = async (match: (m: any) => boolean) => { + for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20)); + return seen.find(match); + }; + const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n"); + const event = (id: number, key: string, body: unknown) => + send({ jsonrpc: "2.0", id, method: "mesh/event", params: { envelope: { key, node: "anchor", body, headers: { "x-event-id": `e${id}`, "x-source": "alpha" } } } }); + try { + // The subscription comes first, from the import, before the bundle has said what it serves. + const sub = await until((m) => m.method === "mesh/subscribe"); + assert.deepEqual(sub.params, { pattern: "alpha.happened" }); + send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } }); + assert.equal(await until((m) => m.id === 1), undefined, "initialize was answered while the import still waited on its subscription"); + send({ jsonrpc: "2.0", id: sub.id, result: {} }); + assert.equal((await until((m) => m.id === 1)).result.serverInfo.name, "watcher"); + send({ jsonrpc: "2.0", id: 2, method: "tools/list" }); + assert.deepEqual((await until((m) => m.id === 2)).result.tools.map((t: any) => t.name), ["relay"]); + + // Handled: answered with a result, so the runtime acknowledges it. + event(10, "alpha.happened", { n: 1 }); + assert.deepEqual((await until((m) => m.id === 10)).result, {}); + assert.match(said, /handled alpha\.happened \{"n":1\}/); + // The handler failed: answered with an error, so the runtime does not acknowledge it. + event(11, "alpha.happened", { fail: true }); + assert.match((await until((m) => m.id === 11)).error.message, /not this one/); + // No handler is for it: answered at once, so it is acknowledged rather than redelivered. + event(12, "beta.other", {}); + assert.deepEqual((await until((m) => m.id === 12)).result, {}); + + // A tool asks the bus: the runtime is asked, and its answer is the tool's. + send({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "relay", arguments: {} } }); + const asked = await until((m) => m.method === "mesh/ask"); + assert.deepEqual(asked.params, { key: "beta.one", body: { x: 1 } }); + send({ jsonrpc: "2.0", id: asked.id, result: { beta: 1 } }); + assert.deepEqual(JSON.parse((await until((m) => m.id === 3)).result.content[0].text), { beta: 1 }); + } finally { + child.kill(); + } +}); + +test("topics match as the runtime's do: * one segment, # and ** any number", async () => { + const { topicMatches } = await import("../dist/stdio/index.js"); + assert.ok(topicMatches("alpha.happened", "alpha.happened")); + assert.ok(topicMatches("alpha.*", "alpha.happened")); + assert.ok(!topicMatches("alpha.*", "alpha.happened.twice")); + assert.ok(topicMatches("#", "anything.at.all")); + assert.ok(topicMatches("alpha.**", "alpha.a.b")); + assert.ok(!topicMatches("beta.*", "alpha.happened")); +});