"""Sync generated TypeScript protocol surfaces from canonical manifests.""" from __future__ import annotations import argparse import json from pathlib import Path from typing import Any from .events import load_event_manifest from .slash_commands import load_slash_command_manifest ROOT = Path(__file__).resolve().parents[2] TUI_PROTOCOL_PATH = ROOT / "apps" / "tui" / "src" / "bridge" / "protocol.ts" TUI_REGISTRY_PATH = ROOT / "apps" / "tui" / "src" / "commands" / "registry.ts" EVENTS_BEGIN = "// " EVENTS_END = "// " EVENT_MAP_BEGIN = "// " EVENT_MAP_END = "// " COMMANDS_BEGIN = "// " COMMANDS_END = "// " def _quote(value: str) -> str: return json.dumps(value) def _ts_string_array(values: list[str], *, indent: str = " ") -> str: if not values: return "[]" lines = ["["] lines.extend(f"{indent}{_quote(value)}," for value in values) lines.append("]") return "\n".join(lines) def _ts_type_union(values: list[str], *, indent: str = " ") -> str: if not values: return "never" return "\n".join(f"{indent}| {_quote(value)}" for value in values) def _unique(values: list[str]) -> list[str]: seen: set[str] = set() result: list[str] = [] for value in values: if value in seen: continue seen.add(value) result.append(value) return result def _replace_section(source: str, begin: str, end: str, rendered: str) -> str: start = source.find(begin) finish = source.find(end) if start == -1 or finish == -1 or finish < start: raise RuntimeError(f"missing generated section markers {begin!r} / {end!r}") finish += len(end) return source[:start] + rendered.rstrip() + source[finish:] def _command_literal(command: dict[str, Any]) -> dict[str, Any]: handler = str(command["handler"]) literal: dict[str, Any] = { "handler": handler, "availability": [handler], "implemented": bool(command.get("implemented", True)), "name": str(command["name"]), "description": str(command["description"]), "summary": str(command["summary"]), "usage": str(command["usage"]), "category": str(command["category"]), } aliases = [str(alias) for alias in command.get("aliases", [])] if aliases: literal["aliases"] = aliases args = command.get("args", []) if args: literal["args"] = [ { "name": str(arg["name"]), "required": bool(arg["required"]), "description": str(arg["description"]), } for arg in args ] if not bool(command.get("tui_visible", True)): literal["isHidden"] = True return literal def render_protocol_events_section() -> str: manifest = load_event_manifest() tui_to_core = [str(value) for value in manifest["tui_to_core"]] core_to_tui = [str(value) for value in manifest["core_to_tui"]] all_events = _unique(tui_to_core + core_to_tui) policy = manifest.get("unknown_event", {}) warning_event = str(policy.get("event") or "notification") warning_level = str(policy.get("level") or "warn") warning_title = str(policy.get("title") or "Protocol warning") return "\n".join( [ EVENTS_BEGIN, "// Generated by python -m openspace.protocol.codegen --write", "export const TUI_TO_CORE_MSG_TYPES = " + _ts_string_array(tui_to_core) + " as const;", "", "export const CORE_TO_TUI_MSG_TYPES = " + _ts_string_array(core_to_tui) + " as const;", "", "export type TuiToCoreMsgType = typeof TUI_TO_CORE_MSG_TYPES[number];", "", "export type CoreToTuiMsgType = typeof CORE_TO_TUI_MSG_TYPES[number];", "", "export type EventType = TuiToCoreMsgType | CoreToTuiMsgType;", "", "const KNOWN_EVENT_TYPES: ReadonlySet = new Set(" + _ts_string_array(all_events) + ");", "", "export function isKnownEventType(value: string): value is EventType {", " return KNOWN_EVENT_TYPES.has(value);", "}", "", "export function makeProtocolWarningMessage(", " eventType: string,", " reason: string,", "): IPCMessage<\"notification\"> {", " return {", f" type: {_quote(warning_event)},", " data: {", f" level: {_quote(warning_level)},", f" title: {_quote(warning_title)},", " message: `${reason}: ${eventType}`,", " event_type: eventType,", " },", " };", "}", EVENTS_END, ] ) def render_protocol_event_map_section() -> str: manifest = load_event_manifest() payload_types = manifest["payload_types"] lines = [ EVENT_MAP_BEGIN, "// Generated by python -m openspace.protocol.codegen --write", "export type EventDataMap = {", ] for event_type in _unique([*manifest["tui_to_core"], *manifest["core_to_tui"]]): lines.append(f" {event_type}: {payload_types[event_type]};") lines.extend(["};", EVENT_MAP_END]) return "\n".join(lines) def render_registry_section() -> str: manifest = load_slash_command_manifest() commands = manifest["commands"] categories = manifest.get("categories", {}) command_names = [str(command["name"]) for command in commands] category_names = [str(name) for name in categories] registry_json = json.dumps( [_command_literal(command) for command in commands], indent=2, ensure_ascii=False, ) category_json = json.dumps(categories, indent=2, ensure_ascii=False) return "\n".join( [ COMMANDS_BEGIN, "// Generated by python -m openspace.protocol.codegen --write", "export type SlashCommandName =", _ts_type_union(command_names) + ";", "", "export type SlashCommandCategory =", _ts_type_union(category_names) + ";", "", "export type SlashCommandArg = {", " name: string;", " required: boolean;", " description: string;", "};", "", "export type SlashCommandDefinition = Command & {", " name: SlashCommandName;", " summary: string;", " usage: string;", " category: SlashCommandCategory;", " args?: SlashCommandArg[];", " hidden?: boolean;", " implemented?: boolean;", "};", "", "export type ParsedSlashCommand = SlashCommandData & {", " args: string[];", " raw: string;", " definition: SlashCommandDefinition | null;", " isSupported: boolean;", "};", "", "const SLASH_COMMAND_REGISTRY: readonly SlashCommandDefinition[] = " + registry_json + " as const;", "", "const CATEGORY_LABELS: Record = " + category_json + ";", COMMANDS_END, ] ) def render_tui_protocol(source: str) -> str: source = _replace_section( source, EVENTS_BEGIN, EVENTS_END, render_protocol_events_section(), ) return _replace_section( source, EVENT_MAP_BEGIN, EVENT_MAP_END, render_protocol_event_map_section(), ) def render_tui_registry(source: str) -> str: return _replace_section( source, COMMANDS_BEGIN, COMMANDS_END, render_registry_section(), ) def sync_generated_files(*, check: bool) -> list[Path]: targets = { TUI_PROTOCOL_PATH: render_tui_protocol(TUI_PROTOCOL_PATH.read_text(encoding="utf-8")), TUI_REGISTRY_PATH: render_tui_registry(TUI_REGISTRY_PATH.read_text(encoding="utf-8")), } changed: list[Path] = [] for path, rendered in targets.items(): current = path.read_text(encoding="utf-8") if current == rendered: continue changed.append(path) if not check: path.write_text(rendered, encoding="utf-8") return changed def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Sync generated TypeScript protocol files from OpenSpace manifests." ) mode = parser.add_mutually_exclusive_group() mode.add_argument("--check", action="store_true", help="fail if generated files are stale") mode.add_argument("--write", action="store_true", help="update generated files") args = parser.parse_args(argv) check = not args.write changed = sync_generated_files(check=check) if not changed: print("protocol generated files are up to date") return 0 for path in changed: print(path.relative_to(ROOT).as_posix()) if check: print("generated files are stale; run python -m openspace.protocol.codegen --write") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())