Merge pull request 'A launched tool emits through its runtime, and calls run beside the stdio loop (hq ADR 0193)' (#14) from feat/0193-a-launched-bundle-emits-through-the-runtime into main

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": {

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<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
+  const ask = (method: string, params: unknown): Promise<unknown> =>
+    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<never> => {
+    throw new Error(`a launched bundle does not ${what} on the bus; its runtime does (novox/hq ADR 0193)`);
+  };
+  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"],
+    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<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 } });
-        }
+        // 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;
       }
       default:

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();

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();
+  }
+});