# Copyright (c) 2026 Storm Dragon # SPDX-License-Identifier: GPL-3.0-or-later """Experimental command/presentation SDK. stdout is reserved for the protocol.""" import json import sys import threading from dataclasses import dataclass, field MAX_FRAME = 64 * 1024 class ProtocolError(Exception): pass @dataclass class Command: id: str title: str description: str handle: object bindings: list = field(default_factory=list) presentation: bool = False def declaration(self): declaration = {"id": self.id, "title": self.title, "description": self.description, "bindings": self.bindings} if self.presentation: declaration["presentation"] = True return declaration def present(speech="", braille="", uninterruptible=False): """Return independent text messages from a presentation-enabled command.""" if type(uninterruptible) is not bool or (uninterruptible and not speech): raise ValueError("Uninterruptible output requires speech and a boolean option") result = {key: text for key, text in (("speech", speech), ("braille", braille)) if text} if uninterruptible: result["uninterruptible"] = True return result def _object(pairs): result = {} for key, value in pairs: if key in result: raise ProtocolError("Duplicate key") result[key] = value return result def _depth(value, depth=0): if depth > 32: raise ProtocolError("JSON nesting limit") if isinstance(value, (dict, list)): for child in value.values() if isinstance(value, dict) else value: _depth(child, depth + 1) def _constant(value): raise ProtocolError("Invalid JSON number") def _encode(value): _depth(value) return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(",", ":")).encode("utf-8") def _fields(value, required, optional=()): if (not isinstance(value, dict) or not set(required) <= value.keys() or value.keys() - set(required) - set(optional)): raise ProtocolError("Invalid fields") def serve(commands, inputStream=None, outputStream=None): """Run in a dedicated child, then exit. Handlers receive (cancelled, arguments). cancelled is a threading.Event shared by the session. Up to eight handlers run concurrently. Shutdown/EOF sets it and joins handlers; the host terminates children that ignore cancellation. Exceptions become generic RPC errors. Streams are binary, owned by this call, and closed on return. """ inputStream = inputStream if inputStream is not None else sys.stdin.buffer outputStream = outputStream if outputStream is not None else sys.stdout.buffer cancelled = threading.Event() writeLock = threading.Lock() workers = [] slots = threading.BoundedSemaphore(8) def write_reply(requestId, value=None, failed=False, release=None): reply = {"jsonrpc": "2.0", "id": requestId} if failed: reply["error"] = {"code": -32603, "message": "Command failed"} else: reply["result"] = value try: frame = _encode(reply) if len(frame) > MAX_FRAME: raise ValueError() except (ValueError, TypeError, UnicodeError, RecursionError, ProtocolError): frame = _encode({"jsonrpc": "2.0", "id": requestId, "error": { "code": -32603, "message": "Invalid or oversized result"}}) with writeLock: # Bound eight handlers/waiting replies plus one serialized writer. if release is not None: release() if not cancelled.is_set(): outputStream.write(frame + b"\n") outputStream.flush() def invoke(requestId, handler, arguments): released = False def release_slot(): nonlocal released if not released: slots.release() released = True try: value = handler(cancelled, arguments) except BaseException: value, failed = None, True else: failed = False try: write_reply(requestId, value, failed, release_slot) except (OSError, ValueError): cancelled.set() finally: release_slot() try: commands = list(commands) handlers = {command.id: command.handle for command in commands} if (not 1 <= len(commands) <= 32 or len(handlers) != len(commands) or any(not key or not callable(value) for key, value in handlers.items())): raise ProtocolError("Invalid commands") session, active, sequence = None, False, 0 while not cancelled.is_set(): frame = inputStream.readline(MAX_FRAME + 2) if not frame: return if not frame.endswith(b"\n") or len(frame) > MAX_FRAME + 1: raise ProtocolError("Invalid frame size or termination") try: message = json.loads(frame.decode("utf-8"), object_pairs_hook=_object, parse_constant=_constant) _depth(message) except (ValueError, UnicodeError, RecursionError) as error: raise ProtocolError("Invalid JSON") from error _fields(message, ("jsonrpc", "method"), ("id", "params")) if message["jsonrpc"] != "2.0": raise ProtocolError("Invalid JSON-RPC version") requestId = message.get("id", "") if requestId: if (not isinstance(requestId, str) or not requestId.isascii() or not requestId.isdecimal() or len(requestId) > 20 or not sequence < int(requestId) < 2**64 or str(int(requestId)) != requestId): raise ProtocolError("Invalid request sequence") sequence = int(requestId) elif "id" in message: raise ProtocolError("Invalid request ID") method, params = message["method"], message.get("params") if method == "initialize": _fields(params, ("protocol", "session", "apis", "capabilities")) version = params["protocol"] if (session is not None or not requestId or version != {"major": 1, "minor": 0} or any(type(value) is not int for value in version.values()) or not isinstance(params["session"], str) or not params["session"] or len(params["session"].encode("utf-8")) > 128 or params["apis"] not in (["commands"], ["commands", "presentation"]) or params["capabilities"] != params["apis"] or (any(command.presentation for command in commands) and params["apis"] != ["commands", "presentation"])): raise ProtocolError("Invalid initialization") session = params["session"] write_reply(requestId, {"protocol": version, "session": session, "commands": [command.declaration() for command in commands]}) elif method == "initialized": _fields(params, ("session",)) if session is None or active or requestId or params["session"] != session: raise ProtocolError("Invalid activation") active = True elif method == "command.invoke": _fields(params, ("command", "arguments")) if (not active or not requestId or not isinstance(params["command"], str) or params["command"] not in handlers or len(_encode(params["arguments"])) > 16 * 1024): raise ProtocolError("Invalid invocation") workers = [worker for worker in workers if worker.is_alive()] if not slots.acquire(blocking=False): raise ProtocolError("Command limit") worker = threading.Thread(target=invoke, args=(requestId, handlers[params["command"]], params["arguments"])) workers.append(worker) worker.start() elif method == "shutdown": if requestId or params is not None: raise ProtocolError("Invalid shutdown") return else: raise ProtocolError("Unknown method") finally: cancelled.set() # Join before closing buffered stdout: a blocked writer must remain # subject to the host deadline rather than race a Python buffer close. for worker in workers: worker.join() inputStream.close() outputStream.close()