diff --git a/package.json b/package.json index 29e8145..fea7f88 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.4", + "version": "0.1.5", "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 5ffb913..4cf4b07 100644 --- a/src/stdio/index.ts +++ b/src/stdio/index.ts @@ -14,6 +14,7 @@ */ import type { ToolDefinition } from "../contracts/index.js"; import { collectTools } from "../tools/index.js"; +import { useBroker, type Broker, type Envelope } from "../messaging/index.js"; /** The protocol version this speaks; the runtime says the same. */ export const PROTOCOL = "2025-03-26"; @@ -44,6 +45,31 @@ export async function serveStdio(name: string, tools: ToolDefinition[]): Promise 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 => + new Promise((resolve, reject) => { + const id = `bundle-${nextOut++}`; + 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)`); + }; + 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"], + close: async () => {}, + }; + useBroker(() => viaRuntime); for await (const line of lines()) { let request: Request; try { @@ -51,6 +77,15 @@ export async function serveStdio(name: string, tools: ToolDefinition[]): Promise } 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; + } const notification = request.id === undefined || request.id === null; switch (request.method) { case "initialize": @@ -74,13 +109,18 @@ export async function serveStdio(name: string, tools: ToolDefinition[]): Promise say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: `${name} has no tool ${wanted}` } }); 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 } }); - } + // 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; } default: diff --git a/test/fixtures/stdio-emits.mjs b/test/fixtures/stdio-emits.mjs new file mode 100644 index 0000000..c61db33 --- /dev/null +++ b/test/fixtures/stdio-emits.mjs @@ -0,0 +1,8 @@ +// A launched bundle whose tool emits (novox/hq ADR 0193): the emit travels to the runtime over stdio. +import { registerModuleTools } from "../../dist/tools/index.js"; +import { emit } from "../../dist/events/index.js"; +import { serveRegisteredOverStdio } from "../../dist/stdio/index.js"; +registerModuleTools("chime", () => [ + { name: "ring", description: "rings and says so", input: {}, run: async () => { await emit("rang", { times: 1 }); return { rang: true }; } }, +]); +await serveRegisteredOverStdio(); diff --git a/test/stdio.test.ts b/test/stdio.test.ts index 32b1b5f..306778e 100644 --- a/test/stdio.test.ts +++ b/test/stdio.test.ts @@ -68,3 +68,39 @@ test("a bundle served as its module lists the seat's verbs as the seat's, whiche child.kill(); } }); + +test("a launched tool's emit is asked of the runtime, and the call answers once the runtime has published (ADR 0193)", async () => { + const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-emits.mjs", import.meta.url))], + { stdio: ["pipe", "pipe", "inherit"], env: { ...process.env, MESH_SERVED_MODULE: "chime", MESH_MODULE: "chime", MESH_NODE: "anchor" } }); + const seen: any[] = []; + let buffered = ""; + 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"); + try { + send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } }); + await until((m) => m.id === 1); + send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "ring", arguments: {} } }); + const asked = await until((m) => m.method === "mesh/publish"); + assert.ok(asked, "the tool's emit was not asked of the runtime"); + assert.equal(asked.params.key, "rang"); + assert.deepEqual(asked.params.body, { times: 1 }); + assert.equal(asked.params.headers["x-source"], "chime"); + assert.equal(seen.find((m) => m.id === 2), undefined, "the call answered before its emit was published"); + send({ jsonrpc: "2.0", id: asked.id, result: {} }); + const answered = await until((m) => m.id === 2); + assert.deepEqual(JSON.parse(answered.result.content[0].text), { rang: true }); + } finally { + child.kill(); + } +});