From 029f9bc53b525d2fe903f4fb5e254646f2c04088 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:39:19 +0800 Subject: [PATCH] feat(extensions): add Pi and OpenClaw sidecar protocol --- nanobot/extensions/__init__.py | 12 + nanobot/extensions/compatibility.py | 214 ++++++++++ nanobot/extensions/node_host.py | 203 ++++++++++ nanobot/extensions/node_sidecar.mjs | 415 ++++++++++++++++++++ nanobot/extensions/protocol.py | 66 ++++ pyproject.toml | 1 + tests/extensions/test_node_compatibility.py | 123 ++++++ 7 files changed, 1034 insertions(+) create mode 100644 nanobot/extensions/compatibility.py create mode 100644 nanobot/extensions/node_host.py create mode 100644 nanobot/extensions/node_sidecar.mjs create mode 100644 nanobot/extensions/protocol.py create mode 100644 tests/extensions/test_node_compatibility.py diff --git a/nanobot/extensions/__init__.py b/nanobot/extensions/__init__.py index 794e1e443..f50d07b5f 100644 --- a/nanobot/extensions/__init__.py +++ b/nanobot/extensions/__init__.py @@ -20,6 +20,13 @@ from nanobot.extensions.manifest import ( ExtensionRuntime, ) from nanobot.extensions.native import discover_native_extensions +from nanobot.extensions.node_host import NodeSidecar +from nanobot.extensions.protocol import ( + NODE_PROTOCOL_VERSION, + NodeLoadResult, + NodeProtocolError, + NodeRegistration, +) from nanobot.extensions.registry import ( ExtensionCandidate, ExtensionDiagnostic, @@ -48,6 +55,11 @@ __all__ = [ "ExtensionSnapshot", "MANIFEST_FILENAME", "ManifestFormatError", + "NODE_PROTOCOL_VERSION", + "NodeLoadResult", + "NodeProtocolError", + "NodeRegistration", + "NodeSidecar", "ResolvedContribution", "build_extension_catalog", "dump_manifest", diff --git a/nanobot/extensions/compatibility.py b/nanobot/extensions/compatibility.py new file mode 100644 index 000000000..57f988796 --- /dev/null +++ b/nanobot/extensions/compatibility.py @@ -0,0 +1,214 @@ +"""Adapters that project Pi/OpenClaw registrations into native nanobot APIs.""" + +from __future__ import annotations + +from dataclasses import asdict, is_dataclass +from typing import Any +from uuid import uuid4 + +from nanobot.agent.hook import ( + AgentHook, + AgentHookContext, + AgentRunHookContext, +) +from nanobot.agent.tools.base import Tool +from nanobot.bus.events import OutboundMessage +from nanobot.command.router import CommandContext, CommandRouter +from nanobot.extensions.node_host import NodeSidecar +from nanobot.extensions.protocol import NodeLoadResult, NodeRegistration + +_EVENT_MAP = { + "pi": { + "before_run": "agent_start", + "after_run": "agent_end", + "before_execute_tool": "tool_call", + "after_execute_tool": "tool_result", + }, + "openclaw": { + "before_run": "before_agent_run", + "after_run": "agent_end", + "before_execute_tool": "before_tool_call", + "after_execute_tool": "after_tool_call", + }, +} + + +class RemoteTool(Tool): + """Native Tool facade whose implementation remains inside a sidecar.""" + + def __init__(self, host: NodeSidecar, registration: NodeRegistration) -> None: + self._host = host + self._registration = registration + + @property + def name(self) -> str: + return self._registration.name + + @property + def description(self) -> str: + return self._registration.description + + @property + def parameters(self) -> dict[str, Any]: + return self._registration.schema or {"type": "object", "properties": {}} + + @property + def read_only(self) -> bool: + return bool((self._registration.metadata or {}).get("readOnly")) + + async def execute(self, **kwargs: Any) -> Any: + result = await self._host.request( + "extension.call", + { + "kind": "tool", + "name": self.name, + "callId": uuid4().hex, + "input": kwargs, + }, + ) + return result.get("text", "") + + +class RemoteHook(AgentHook): + """Observation-only lifecycle bridge for compatible extension events.""" + + def __init__(self, host: NodeSidecar, runtime: str, events: set[str]) -> None: + super().__init__() + self._host = host + self._events = events + self._mapping = _EVENT_MAP[runtime] + + async def _emit(self, lifecycle: str, context: object, **extra: Any) -> None: + event = self._mapping[lifecycle] + if event not in self._events: + return + payload = _jsonable(context) + if isinstance(payload, dict): + payload.update(extra) + await self._host.request( + "extension.event", + {"name": event, "event": payload}, + ) + + async def before_run(self, context: AgentRunHookContext) -> None: + await self._emit("before_run", context) + + async def after_run(self, context: AgentRunHookContext) -> None: + await self._emit("after_run", context) + + async def before_execute_tool( + self, + context: AgentHookContext, + tool_call: Any, + tool: Any, + params: Any, + ) -> None: + await self._emit( + "before_execute_tool", + context, + toolCall=_jsonable(tool_call), + tool=getattr(tool, "name", ""), + input=_jsonable(params), + ) + + async def after_execute_tool( + self, + context: AgentHookContext, + tool_call: Any, + tool: Any, + params: Any, + result: Any, + ) -> None: + await self._emit( + "after_execute_tool", + context, + toolCall=_jsonable(tool_call), + tool=getattr(tool, "name", ""), + input=_jsonable(params), + result=_jsonable(result), + ) + + +class CompatibleExtension: + """Loaded Pi/OpenClaw extension and its native projections.""" + + def __init__( + self, + *, + host: NodeSidecar, + runtime: str, + owner: str, + result: NodeLoadResult, + ) -> None: + self.host = host + self.runtime = runtime + self.owner = owner + self.result = result + + @property + def tools(self) -> tuple[RemoteTool, ...]: + return tuple( + RemoteTool(self.host, item) + for item in self.result.registrations + if item.kind == "tool" + ) + + @property + def hook(self) -> RemoteHook | None: + events = { + item.name + for item in self.result.registrations + if item.kind == "hook" + } + return RemoteHook(self.host, self.runtime, events) if events else None + + def register_commands(self, router: CommandRouter) -> None: + for item in self.result.registrations: + if item.kind != "command": + continue + + async def handler(ctx: CommandContext, name: str = item.name) -> OutboundMessage | None: + result = await self.host.request( + "extension.call", + { + "kind": "command", + "name": name, + "input": { + "args": ctx.args, + "raw": ctx.raw, + "channel": ctx.msg.channel, + "chatId": ctx.msg.chat_id, + "senderId": ctx.msg.sender_id, + "sessionKey": ctx.key, + }, + }, + ) + text = str(result.get("text") or "") + if not text: + return None + return OutboundMessage( + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, + content=text, + ) + + command = f"/{item.name}" + router.exact(command, handler, owner=self.owner) + router.prefix(f"{command} ", handler, owner=self.owner) + + async def close(self) -> None: + await self.host.close() + + +def _jsonable(value: Any) -> Any: + if is_dataclass(value): + return _jsonable(asdict(value)) + if isinstance(value, dict): + return {str(key): _jsonable(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(item) for item in value] + if isinstance(value, BaseException): + return {"type": type(value).__name__, "message": str(value)} + if value is None or isinstance(value, (str, int, float, bool)): + return value + return str(value) diff --git a/nanobot/extensions/node_host.py b/nanobot/extensions/node_host.py new file mode 100644 index 000000000..ff09a150b --- /dev/null +++ b/nanobot/extensions/node_host.py @@ -0,0 +1,203 @@ +"""Async process boundary for untrusted-compatible JavaScript extension APIs.""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +from contextlib import suppress +from pathlib import Path +from typing import Any + +from nanobot.extensions.protocol import ( + NODE_PROTOCOL_VERSION, + NodeLoadResult, + NodeProtocolError, +) + + +class NodeSidecar: + """One isolated Node process hosting one extension module.""" + + def __init__( + self, + *, + node: str | None = None, + timeout: float = 30.0, + ) -> None: + self._node = node or os.getenv("NANOBOT_NODE") or shutil.which("node") + self._timeout = timeout + self._process: asyncio.subprocess.Process | None = None + self._reader: asyncio.Task[None] | None = None + self._stderr_reader: asyncio.Task[None] | None = None + self._pending: dict[int, asyncio.Future[Any]] = {} + self._write_lock = asyncio.Lock() + self._next_id = 0 + self.stderr: list[str] = [] + + @property + def running(self) -> bool: + return self._process is not None and self._process.returncode is None + + async def start(self) -> None: + if self.running: + return + if not self._node: + raise NodeProtocolError( + "Node.js is required for Pi and OpenClaw extensions; " + "install Node.js 20+ or set NANOBOT_NODE" + ) + script = Path(__file__).with_name("node_sidecar.mjs") + self._process = await asyncio.create_subprocess_exec( + self._node, + str(script), + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + self._reader = asyncio.create_task(self._read_stdout()) + self._stderr_reader = asyncio.create_task(self._read_stderr()) + hello = await self.request("hello", {}) + if hello.get("protocol") != NODE_PROTOCOL_VERSION: + await self.close() + raise NodeProtocolError( + f"sidecar protocol mismatch: expected {NODE_PROTOCOL_VERSION}" + ) + + async def load( + self, + *, + runtime: str, + entry: Path, + extension_id: str, + name: str, + version: str, + config: dict[str, Any] | None = None, + workspace: Path | None = None, + ) -> NodeLoadResult: + await self.start() + result = await self.request( + "extension.load", + { + "runtime": runtime, + "entry": str(entry.expanduser().resolve()), + "identity": { + "id": extension_id, + "name": name, + "version": version, + }, + "config": config or {}, + "workspace": str((workspace or Path.cwd()).resolve()), + }, + ) + return NodeLoadResult.from_mapping(result) + + async def request( + self, + method: str, + params: dict[str, Any], + *, + timeout: float | None = None, + ) -> dict[str, Any]: + if method != "hello" and not self.running: + raise NodeProtocolError("sidecar is not running") + process = self._process + if process is None or process.stdin is None: + raise NodeProtocolError("sidecar failed to start") + + self._next_id += 1 + request_id = self._next_id + future = asyncio.get_running_loop().create_future() + self._pending[request_id] = future + message = json.dumps( + { + "protocol": NODE_PROTOCOL_VERSION, + "id": request_id, + "method": method, + "params": params, + }, + separators=(",", ":"), + ).encode() + try: + async with self._write_lock: + process.stdin.write(message + b"\n") + await process.stdin.drain() + result = await asyncio.wait_for(future, timeout or self._timeout) + except Exception: + self._pending.pop(request_id, None) + raise + if not isinstance(result, dict): + raise NodeProtocolError(f"sidecar method {method!r} returned a non-object result") + return result + + async def close(self) -> None: + process = self._process + if process is None: + return + if process.returncode is None: + with suppress(Exception): + await self.request("shutdown", {}, timeout=2.0) + if process.returncode is None: + process.terminate() + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(process.wait(), 2.0) + if process.returncode is None: + process.kill() + await process.wait() + for task in (self._reader, self._stderr_reader): + if task is not None: + task.cancel() + with suppress(asyncio.CancelledError): + await task + self._fail_pending(NodeProtocolError("sidecar closed")) + self._process = None + + async def _read_stdout(self) -> None: + assert self._process and self._process.stdout + try: + while line := await self._process.stdout.readline(): + try: + message = json.loads(line) + request_id = message["id"] + future = self._pending.pop(request_id) + if error := message.get("error"): + future.set_exception( + NodeProtocolError( + f"{error.get('code', 'sidecar_error')}: " + f"{error.get('message', 'unknown sidecar error')}" + ) + ) + else: + future.set_result(message.get("result", {})) + except Exception as exc: + self._fail_pending(NodeProtocolError(f"invalid sidecar response: {exc}")) + finally: + if self._process and self._process.returncode is None: + await self._process.wait() + code = self._process.returncode if self._process else "unknown" + details = self.stderr[-1] if self.stderr else "no diagnostics" + self._fail_pending( + NodeProtocolError(f"sidecar exited with code {code}: {details}") + ) + + async def _read_stderr(self) -> None: + assert self._process and self._process.stderr + while line := await self._process.stderr.readline(): + text = line.decode(errors="replace").rstrip() + if text: + self.stderr.append(text) + del self.stderr[:-100] + + def _fail_pending(self, error: Exception) -> None: + for future in self._pending.values(): + if not future.done(): + future.set_exception(error) + self._pending.clear() + + async def __aenter__(self) -> NodeSidecar: + await self.start() + return self + + async def __aexit__(self, *_: object) -> None: + await self.close() diff --git a/nanobot/extensions/node_sidecar.mjs b/nanobot/extensions/node_sidecar.mjs new file mode 100644 index 000000000..348d66214 --- /dev/null +++ b/nanobot/extensions/node_sidecar.mjs @@ -0,0 +1,415 @@ +import { AsyncLocalStorage } from "node:async_hooks"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import readline from "node:readline"; + +const PROTOCOL = 1; +const invocations = new AsyncLocalStorage(); +const state = { + runtime: null, + identity: null, + workspace: process.cwd(), + config: {}, + tools: new Map(), + commands: new Map(), + hooks: new Map(), + registrations: [], + diagnostics: [], +}; + +const writeError = (...parts) => process.stderr.write(`${parts.map(String).join(" ")}\n`); +console.log = writeError; +console.info = writeError; +console.warn = writeError; +console.error = writeError; + +function sanitize(value, depth = 0, seen = new WeakSet()) { + if (depth > 12) return "[max depth]"; + if (value === null || ["string", "number", "boolean"].includes(typeof value)) return value; + if (typeof value === "bigint") return String(value); + if (typeof value === "function" || typeof value === "undefined") return undefined; + if (typeof value !== "object") return String(value); + if (seen.has(value)) return "[circular]"; + seen.add(value); + if (Array.isArray(value)) return value.map((item) => sanitize(item, depth + 1, seen)); + return Object.fromEntries( + Object.entries(value) + .map(([key, item]) => [key, sanitize(item, depth + 1, seen)]) + .filter(([, item]) => item !== undefined), + ); +} + +function addRegistration(kind, name, options = {}) { + const registration = { + kind, + name: String(name), + description: String(options.description || ""), + ...(options.schema ? { schema: sanitize(options.schema) } : {}), + ...(options.metadata ? { metadata: sanitize(options.metadata) } : {}), + }; + const index = state.registrations.findIndex( + (item) => item.kind === registration.kind && item.name === registration.name, + ); + if (index >= 0) state.registrations[index] = registration; + else state.registrations.push(registration); +} + +function unsupported(name) { + if (!state.diagnostics.includes(`Unsupported compatibility API: ${name}`)) { + state.diagnostics.push(`Unsupported compatibility API: ${name}`); + } +} + +function unsupportedFacade(path) { + const fn = () => { + throw new Error(`${path} is not available in the nanobot compatibility host`); + }; + return new Proxy(fn, { + get: (_, key) => unsupportedFacade(`${path}.${String(key)}`), + }); +} + +function addHook(name, handler, flavor) { + const names = Array.isArray(name) ? name : [name]; + for (const item of names) { + const key = String(item); + const handlers = state.hooks.get(key) || []; + handlers.push({ handler, flavor }); + state.hooks.set(key, handlers); + addRegistration("hook", key); + } +} + +function addTool(tool, flavor, options = {}) { + if (typeof tool === "function") { + const resolved = tool({ + config: state.config, + runtimeConfig: state.config, + getRuntimeConfig: () => state.config, + workspaceDir: state.workspace, + sandboxed: true, + }); + for (const item of Array.isArray(resolved) ? resolved : [resolved]) { + if (item) addTool(item, flavor, options); + } + return; + } + if (!tool || typeof tool !== "object" || typeof tool.name !== "string") { + throw new Error("registered tool must define a name"); + } + state.tools.set(tool.name, { tool, flavor }); + addRegistration("tool", tool.name, { + description: tool.description, + schema: tool.parameters || { type: "object", properties: {} }, + metadata: { + label: tool.label, + optional: options.optional === true, + readOnly: tool.readOnly === true, + }, + }); +} + +function invocationOutput(value) { + const context = invocations.getStore(); + if (context) context.outputs.push(value); +} + +function eventBus() { + return { + on: (name, handler) => addHook(`event:${name}`, handler, "pi-event"), + emit: async (name, data) => emitEvent(`event:${name}`, data), + }; +} + +function piApi() { + const api = { + on: (name, handler) => addHook(name, handler, "pi"), + registerTool: (tool) => addTool(tool, "pi"), + registerCommand: (name, options) => { + state.commands.set(String(name), { command: options, flavor: "pi" }); + addRegistration("command", name, { description: options?.description }); + }, + registerProvider: (nameOrProvider, config) => { + const provider = + typeof nameOrProvider === "string" + ? { id: nameOrProvider, ...config } + : nameOrProvider; + addRegistration("llm_provider", provider.id || provider.name, { + description: provider.name, + metadata: provider, + }); + }, + unregisterProvider: () => {}, + sendMessage: (message) => invocationOutput(message?.content || message), + sendUserMessage: (message) => invocationOutput(message), + appendEntry: () => unsupported("pi.appendEntry"), + setSessionName: () => unsupported("pi.setSessionName"), + getSessionName: () => undefined, + setLabel: () => unsupported("pi.setLabel"), + getActiveTools: () => [], + getAllTools: () => [], + setActiveTools: () => unsupported("pi.setActiveTools"), + getCommands: () => [], + registerShortcut: () => unsupported("pi.registerShortcut"), + registerFlag: () => unsupported("pi.registerFlag"), + getFlag: () => undefined, + registerMessageRenderer: () => unsupported("pi.registerMessageRenderer"), + registerEntryRenderer: () => unsupported("pi.registerEntryRenderer"), + exec: unsupportedFacade("pi.exec"), + setModel: async () => false, + getThinkingLevel: () => "off", + setThinkingLevel: () => unsupported("pi.setThinkingLevel"), + events: eventBus(), + }; + return new Proxy(api, { + get(target, key) { + if (key in target) return target[key]; + unsupported(`pi.${String(key)}`); + return unsupportedFacade(`pi.${String(key)}`); + }, + }); +} + +function openClawApi() { + const identity = state.identity; + const api = { + id: identity.id, + name: identity.name, + version: identity.version, + source: state.entry, + rootDir: state.rootDir, + registrationMode: "activate", + config: state.config, + pluginConfig: state.config, + runtime: unsupportedFacade("openclaw.runtime"), + logger: { + debug: writeError, + info: writeError, + warn: writeError, + error: writeError, + }, + registerTool: (tool, options) => addTool(tool, "openclaw", options), + registerCommand: (command) => { + state.commands.set(command.name, { command, flavor: "openclaw" }); + addRegistration("command", command.name, { description: command.description }); + }, + registerHook: (names, handler) => addHook(names, handler, "openclaw"), + on: (name, handler) => addHook(name, handler, "openclaw"), + registerProvider: (provider) => + addRegistration("llm_provider", provider.id, { + description: provider.label, + metadata: provider, + }), + registerRealtimeTranscriptionProvider: (provider) => + addRegistration("transcription_provider", provider.id, { + description: provider.label, + metadata: provider, + }), + registerImageGenerationProvider: (provider) => + addRegistration("image_generation_provider", provider.id, { + description: provider.label, + metadata: provider, + }), + registerWebSearchProvider: (provider) => + addRegistration("web_search_provider", provider.id, { + description: provider.label, + metadata: provider, + }), + resolvePath: (value) => new URL(value, pathToFileURL(`${state.rootDir}/`)).pathname, + }; + const grouped = unsupportedFacade("openclaw"); + api.session = grouped.session; + api.agent = grouped.agent; + api.runContext = grouped.runContext; + api.lifecycle = grouped.lifecycle; + return new Proxy(api, { + get(target, key) { + if (key in target) return target[key]; + if (String(key).startsWith("register")) unsupported(`openclaw.${String(key)}`); + return (..._args) => unsupported(`openclaw.${String(key)}`); + }, + }); +} + +function unwrapModule(value) { + const seen = new Set(); + let current = value; + for (let index = 0; index < 12 && current && !seen.has(current); index += 1) { + seen.add(current); + if (typeof current === "function" || typeof current?.register === "function") return current; + current = current.default ?? current.module; + } + return current; +} + +async function importModule(entry) { + try { + return await import(`${pathToFileURL(entry).href}?nanobot=${Date.now()}`); + } catch (error) { + if (![".ts", ".tsx", ".cts", ".mts"].some((suffix) => entry.endsWith(suffix))) throw error; + try { + const imported = await import("jiti"); + const createJiti = imported.createJiti || imported.default; + return await createJiti(import.meta.url, { interopDefault: true }).import(entry); + } catch (jitiError) { + throw new Error( + `Could not load TypeScript extension. Use Node.js with type stripping or install jiti. ${jitiError.message}`, + { cause: error }, + ); + } + } +} + +async function loadExtension(params) { + if (!["pi", "openclaw"].includes(params.runtime)) { + throw new Error(`unsupported Node extension runtime: ${params.runtime}`); + } + state.runtime = params.runtime; + state.identity = params.identity; + state.workspace = params.workspace; + state.config = params.config || {}; + state.entry = params.entry; + state.rootDir = fileURLToPath(new URL(".", pathToFileURL(params.entry))); + state.tools.clear(); + state.commands.clear(); + state.hooks.clear(); + state.registrations.length = 0; + state.diagnostics.length = 0; + + const loaded = unwrapModule(await importModule(params.entry)); + const factory = + params.runtime === "openclaw" && typeof loaded?.register === "function" + ? loaded.register + : loaded; + if (typeof factory !== "function") throw new Error("extension entry does not export a factory"); + const result = factory(params.runtime === "pi" ? piApi() : openClawApi()); + if (params.runtime === "openclaw" && result?.then) { + throw new Error("OpenClaw plugin register must be synchronous"); + } + if (params.runtime === "pi") await result; + return { + registrations: state.registrations, + diagnostics: state.diagnostics, + }; +} + +function contextApi() { + return { + mode: "rpc", + hasUI: false, + cwd: state.workspace, + signal: undefined, + ui: new Proxy( + { notify: (message) => invocationOutput(message) }, + { get: (target, key) => target[key] || unsupportedFacade(`pi.ui.${String(key)}`) }, + ), + isIdle: () => true, + isProjectTrusted: () => true, + hasPendingMessages: () => false, + getContextUsage: () => undefined, + getSystemPrompt: () => "", + }; +} + +function resultText(result, outputs = []) { + const values = [...outputs]; + if (result !== undefined) values.push(result); + const text = []; + for (const value of values) { + if (typeof value === "string") text.push(value); + else if (typeof value?.text === "string") text.push(value.text); + else if (typeof value?.content === "string") text.push(value.content); + else if (Array.isArray(value?.content)) { + for (const item of value.content) { + if (typeof item === "string") text.push(item); + else if (typeof item?.text === "string") text.push(item.text); + } + } else if (value !== undefined) text.push(JSON.stringify(sanitize(value))); + } + return text.filter(Boolean).join("\n"); +} + +async function callExtension(params) { + const context = { outputs: [] }; + return invocations.run(context, async () => { + if (params.kind === "tool") { + const record = state.tools.get(params.name); + if (!record) throw new Error(`unknown tool: ${params.name}`); + const result = await record.tool.execute( + params.callId || "nanobot", + params.input || {}, + undefined, + undefined, + ...(record.flavor === "pi" ? [contextApi()] : []), + ); + return { text: resultText(result, context.outputs), raw: sanitize(result) }; + } + if (params.kind === "command") { + const record = state.commands.get(params.name); + if (!record) throw new Error(`unknown command: ${params.name}`); + const input = params.input || {}; + const result = + record.flavor === "pi" + ? await record.command.handler(input.args || "", contextApi()) + : await record.command.handler({ + args: input.args || "", + commandBody: input.raw || `/${params.name}`, + channel: input.channel || "websocket", + senderId: input.senderId, + isAuthorizedSender: true, + config: state.config, + sessionKey: input.sessionKey, + requestConversationBinding: async () => ({ ok: false }), + detachConversationBinding: async () => ({ removed: false }), + getCurrentConversationBinding: async () => null, + }); + return { text: resultText(result, context.outputs), raw: sanitize(result) }; + } + throw new Error(`unsupported callable kind: ${params.kind}`); + }); +} + +async function emitEvent(name, event) { + const handlers = state.hooks.get(name) || []; + const results = []; + for (const { handler, flavor } of handlers) { + results.push( + await handler(event, flavor === "pi" ? contextApi() : { config: state.config }), + ); + } + return { results: sanitize(results) }; +} + +async function dispatch(method, params) { + if (method === "hello") { + return { protocol: PROTOCOL, node: process.version }; + } + if (method === "extension.load") return loadExtension(params); + if (method === "extension.call") return callExtension(params); + if (method === "extension.event") return emitEvent(params.name, params.event); + if (method === "shutdown") { + queueMicrotask(() => process.exit(0)); + return {}; + } + throw new Error(`unknown method: ${method}`); +} + +let queue = Promise.resolve(); +const lines = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); +lines.on("line", (line) => { + queue = queue.then(async () => { + let request; + try { + request = JSON.parse(line); + if (request.protocol !== PROTOCOL) throw new Error("protocol version mismatch"); + const result = await dispatch(request.method, request.params || {}); + process.stdout.write(`${JSON.stringify({ id: request.id, result: sanitize(result) })}\n`); + } catch (error) { + process.stdout.write( + `${JSON.stringify({ + id: request?.id ?? null, + error: { code: "extension_error", message: String(error?.message || error) }, + })}\n`, + ); + } + }); +}); diff --git a/nanobot/extensions/protocol.py b/nanobot/extensions/protocol.py new file mode 100644 index 000000000..d8290d0ec --- /dev/null +++ b/nanobot/extensions/protocol.py @@ -0,0 +1,66 @@ +"""Versioned messages shared with the Node compatibility sidecar.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +NODE_PROTOCOL_VERSION = 1 + + +class NodeProtocolError(RuntimeError): + """The sidecar returned an invalid message or reported an RPC failure.""" + + +@dataclass(frozen=True, slots=True) +class NodeRegistration: + """One callable or inspectable contribution retained by the sidecar.""" + + kind: str + name: str + description: str = "" + schema: dict[str, Any] | None = None + metadata: dict[str, Any] | None = None + + @classmethod + def from_mapping(cls, value: object) -> NodeRegistration: + if not isinstance(value, dict): + raise NodeProtocolError("sidecar registration must be an object") + kind = value.get("kind") + name = value.get("name") + if not isinstance(kind, str) or not isinstance(name, str): + raise NodeProtocolError("sidecar registration requires string kind and name") + schema = value.get("schema") + metadata = value.get("metadata") + if schema is not None and not isinstance(schema, dict): + raise NodeProtocolError("sidecar registration schema must be an object") + if metadata is not None and not isinstance(metadata, dict): + raise NodeProtocolError("sidecar registration metadata must be an object") + return cls( + kind=kind, + name=name, + description=str(value.get("description") or ""), + schema=schema, + metadata=metadata, + ) + + +@dataclass(frozen=True, slots=True) +class NodeLoadResult: + """Metadata returned after a Pi or OpenClaw module registers itself.""" + + registrations: tuple[NodeRegistration, ...] + diagnostics: tuple[str, ...] = () + + @classmethod + def from_mapping(cls, value: object) -> NodeLoadResult: + if not isinstance(value, dict): + raise NodeProtocolError("sidecar load result must be an object") + registrations = value.get("registrations", []) + diagnostics = value.get("diagnostics", []) + if not isinstance(registrations, list) or not isinstance(diagnostics, list): + raise NodeProtocolError("invalid sidecar load result") + return cls( + registrations=tuple(NodeRegistration.from_mapping(item) for item in registrations), + diagnostics=tuple(str(item) for item in diagnostics), + ) diff --git a/pyproject.toml b/pyproject.toml index da5dede45..f37757e0a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -122,6 +122,7 @@ allow-direct-references = true [tool.hatch.build] include = [ "nanobot/**/*.py", + "nanobot/extensions/**/*.mjs", "nanobot/templates/**/*.md", "nanobot/skills/**/*.md", "nanobot/skills/**/*.sh", diff --git a/tests/extensions/test_node_compatibility.py b/tests/extensions/test_node_compatibility.py new file mode 100644 index 000000000..e0a734de0 --- /dev/null +++ b/tests/extensions/test_node_compatibility.py @@ -0,0 +1,123 @@ +from pathlib import Path + +import pytest + +from nanobot.extensions.compatibility import CompatibleExtension +from nanobot.extensions.node_host import NodeSidecar + + +def _write(path: Path, content: str) -> Path: + path.write_text(content) + return path + + +@pytest.mark.asyncio +async def test_pi_extension_loads_tools_commands_and_events(tmp_path: Path) -> None: + entry = _write( + tmp_path / "pi-extension.mjs", + """ + export default function (pi) { + pi.registerTool({ + name: "pi_echo", + label: "Echo", + description: "Echo input", + parameters: { + type: "object", + properties: { text: { type: "string" } }, + required: ["text"] + }, + async execute(_id, params) { + return { content: [{ type: "text", text: `pi:${params.text}` }] }; + } + }); + pi.registerCommand("hello", { + description: "Say hello", + async handler(args, ctx) { ctx.ui.notify(`hello:${args}`); } + }); + pi.on("agent_start", (event) => { + if (!event.messages) throw new Error("missing messages"); + }); + } + """, + ) + host = NodeSidecar() + try: + result = await host.load( + runtime="pi", + entry=entry, + extension_id="test.pi", + name="Pi test", + version="1.0.0", + workspace=tmp_path, + ) + extension = CompatibleExtension( + host=host, + runtime="pi", + owner="test.pi", + result=result, + ) + assert [(item.kind, item.name) for item in result.registrations] == [ + ("tool", "pi_echo"), + ("command", "hello"), + ("hook", "agent_start"), + ] + assert await extension.tools[0].execute(text="ok") == "pi:ok" + assert extension.hook is not None + finally: + await host.close() + + +@pytest.mark.asyncio +async def test_openclaw_definition_loads_and_invokes_tool(tmp_path: Path) -> None: + entry = _write( + tmp_path / "openclaw-plugin.cjs", + """ + module.exports = { + id: "test.openclaw", + register(api) { + api.registerTool({ + name: "claw_echo", + description: "Echo input", + parameters: { + type: "object", + properties: { text: { type: "string" } }, + required: ["text"] + }, + async execute(_id, params) { + return { content: [{ type: "text", text: `claw:${params.text}` }] }; + } + }); + api.registerCommand({ + name: "status", + description: "Show status", + handler: async () => ({ text: "ready" }) + }); + api.on("agent_end", () => undefined); + } + }; + """, + ) + host = NodeSidecar() + try: + result = await host.load( + runtime="openclaw", + entry=entry, + extension_id="test.openclaw", + name="OpenClaw test", + version="1.0.0", + workspace=tmp_path, + ) + extension = CompatibleExtension( + host=host, + runtime="openclaw", + owner="test.openclaw", + result=result, + ) + assert await extension.tools[0].execute(text="ok") == "claw:ok" + assert {item.name for item in result.registrations} == { + "claw_echo", + "status", + "agent_end", + } + finally: + await host.close()