vdm/contracts/codegen/gen_openrpc.py

#!/usr/bin/env python3
"""Emit contracts/openrpc.json from contracts/schema/.

This is the document humans read. It is generated, not written, so it cannot drift from
the schemas the code is generated from — the failure mode where the docs say one thing and
the wire does another is designed out rather than policed.

JSON-RPC named parameters are modelled as OpenRPC `by-name` params: each property of a
method's params object becomes one entry, which is what a reader expects to see.

Server-to-client notifications are not expressible in OpenRPC 1.2, so they are emitted
under a top-level `x-events` key alongside their payload schemas.

Run: python3 contracts/codegen/gen_openrpc.py
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from schema_ir import ID_PREFIX, SCHEMA_ROOT, Loader  # noqa: E402

ROOT = SCHEMA_ROOT.parent
OUT = ROOT / "openrpc.json"

STRIP = {"$schema", "$id", "title"}


def rewrite(node: object) -> object:
    """Point every $ref at #/components/schemas/<Name> and drop per-file keywords."""
    if isinstance(node, list):
        return [rewrite(n) for n in node]
    if not isinstance(node, dict):
        return node
    out: dict[str, object] = {}
    for k, v in node.items():
        if k == "$ref" and isinstance(v, str):
            if v.startswith("#/$defs/"):
                # envelope.schema.json's internal refs land under the x-envelope key.
                out["$ref"] = "#/x-envelope/" + v[len("#/$defs/"):]
                continue
            if not v.startswith(ID_PREFIX + "types/"):
                raise SystemExit(f"openrpc: unexpected $ref target {v}")
            name = v[len(ID_PREFIX + "types/"):].removesuffix(".schema.json")
            out["$ref"] = f"#/components/schemas/{name}"
            continue
        if k in STRIP:
            continue
        out[k] = rewrite(v)
    return out


def main() -> int:
    loader = Loader()
    version = (ROOT / "VERSION").read_text().strip()

    schemas: dict[str, object] = {}
    for sid, doc in sorted(loader.by_id.items()):
        if "/types/" not in sid:
            continue
        name = doc["title"]
        body = rewrite({k: v for k, v in doc.items() if k not in STRIP})
        assert isinstance(body, dict)
        body["title"] = name
        schemas[name] = body

    methods = []
    for sid, doc in sorted(loader.by_id.items()):
        if "/methods/" not in sid:
            continue
        params_schema = doc["properties"]["params"]

        # Expand a params object into by-name entries. A $ref'd params object is resolved
        # first so the reader sees the fields, not just a type name.
        resolved = params_schema
        if "$ref" in resolved:
            resolved = loader.by_id[resolved["$ref"]]
        params = []
        required = set(resolved.get("required", []))
        for prop, sub in resolved.get("properties", {}).items():
            entry: dict[str, object] = {"name": prop, "schema": rewrite(sub)}
            if prop in required:
                entry["required"] = True
            if isinstance(sub, dict) and sub.get("description"):
                entry["description"] = sub["description"]
            params.append(entry)

        method: dict[str, object] = {
            "name": doc["title"],
            "summary": doc.get("description", "").split(".")[0] + ".",
            "description": doc.get("description", ""),
            "paramStructure": "by-name",
            "params": params,
            "result": {"name": f"{doc['title']}Result", "schema": rewrite(doc["properties"]["result"])},
            "x-privileged": doc.get("x-privileged", False),
            "x-transports": doc.get("x-transports", []),
            "x-deadlineMs": doc.get("x-deadlineMs"),
        }
        if doc.get("x-errors"):
            code_doc = {e["value"]: e["doc"] for e in
                        loader.by_id[ID_PREFIX + "types/ErrorCode.schema.json"]["x-enum"]}
            method["errors"] = [{"code": c, "message": code_doc.get(c, "")} for c in doc["x-errors"]]
        if doc.get("x-wsRestrictions"):
            method["x-wsRestrictions"] = doc["x-wsRestrictions"]
        methods.append(method)

    events = []
    for sid, doc in sorted(loader.by_id.items()):
        if "/events/" not in sid:
            continue
        events.append({
            "name": doc["title"],
            "description": doc.get("description", ""),
            "params": rewrite(doc["properties"]["params"]),
            "x-maxRateHz": doc.get("x-maxRateHz"),
        })

    envelope = loader.by_id[ID_PREFIX + "envelope.schema.json"]

    out = {
        "openrpc": "1.2.6",
        "info": {
            "title": "Velox Download Manager",
            "version": version,
            "description": (
                "The wire contract between veloxd and every client: the Qt GUI, the CLI, "
                "the native-messaging host and the Firefox extension. One JSON-RPC 2.0 "
                "payload set over four framings; only the framing differs.\n\n"
                "GENERATED from contracts/schema/ by contracts/codegen/gen_openrpc.py. "
                "Do not edit by hand."
            ),
            "license": {"name": "See repository LICENSE"},
        },
        "servers": [
            {"name": "unix-socket", "url": "unix:$XDG_RUNTIME_DIR/velox/velox.sock",
             "description": "NDJSON. GUI, CLI and nmhost. Peer credentials checked via SO_PEERCRED; same UID only, no token."},
            {"name": "loopback-ws", "url": "ws://127.0.0.1:52000",
             "description": "One JSON message per text frame. Extension fallback. Bound to 127.0.0.1 only, Origin-checked, token-authenticated, rate-limited. Port is the first free one in 52000-52016."},
        ],
        "methods": methods,
        "components": {"schemas": schemas},
        "x-events": events,
        "x-envelope": rewrite(envelope.get("$defs", {})),
        "x-transports": {
            "uds": "Unix domain socket, newline-delimited JSON.",
            "ws": "Loopback WebSocket, one JSON message per text frame. Privileged methods are refused here with -32003.",
        },
    }

    OUT.write_text(json.dumps(out, indent=2) + "\n")
    print(f"gen_openrpc: {len(methods)} methods, {len(events)} events, {len(schemas)} schemas -> {OUT}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())