e3a53a9373
./scripts/<name>.py now sources activate when the inherited environment is missing or stale (scripts/_hermes-python) and runs under the pinned interpreter, so these work from a bare shell and any cwd. The fast path from an activated shell costs ~5ms over bare python. These all import repo or locked third-party code. CI invokes them as `python3 scripts/<name>.py` under setup-pm's environment, where the shebang is inert, so no require_activation() guard: that would fail those lanes, which run without __HERMES_ACTIVATED. Left out on purpose: keystroke_diagnostic.py (run by users inside their own install, where activation would provision their home) and docker_config_migrate.py (runs in the container through its venv python).
368 lines
15 KiB
Python
Executable File
368 lines
15 KiB
Python
Executable File
#!/usr/bin/env -S bash -c 'exec "$BASH" "$(dirname "$0")/_hermes-python" "$0" "$@"'
|
|
"""Render ``tui_gateway/contracts`` into TypeScript and OpenRPC.
|
|
|
|
Python-only (the Python CI lane has no Node): Pydantic's ``model_json_schema()`` output is walked
|
|
by a deliberately small JSON-Schema-subset renderer — object/properties/required, primitives,
|
|
enum, const, anyOf-with-null, array/items, ``$ref``, oneOf + discriminator, additionalProperties.
|
|
Anything else raises at generation time so an unsupported model is fixed at the model, never
|
|
worked around in the output. Prettier runs on the TS when a node_modules binary is present
|
|
(output is already in the repo's prettier style; the Python CI lane regenerates and diffs it).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import OrderedDict
|
|
from pathlib import Path
|
|
from typing import Any, get_args, get_type_hints
|
|
|
|
from pydantic import TypeAdapter
|
|
from pydantic.json_schema import GenerateJsonSchema
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from tui_gateway import contracts # noqa: E402,F401 (imports every topic module → fills the tables)
|
|
from tui_gateway.contracts.connectors import ( # noqa: E402
|
|
ConnectorAccountStatus,
|
|
ConnectorErrorReason,
|
|
ConnectorToolFacet,
|
|
ConnectorToolsSource,
|
|
)
|
|
from tui_gateway.contracts.connectors_operation import ConnectionSettleReason, ConnectionTargetState # noqa: E402
|
|
from tui_gateway.contracts.registry import EVENTS, METHODS, SERVER_REQUESTS # noqa: E402
|
|
from tools.connectors.contract import SettleReason, TargetState # noqa: E402
|
|
from tools.connectors.gateway.wire import ConnectionStatus # noqa: E402
|
|
from tools.connectors.portal.tools_cache import ToolsRead # noqa: E402
|
|
from tools.connectors.portal.wire import ConnectorTool # noqa: E402
|
|
|
|
TS_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.generated.ts"
|
|
OPENRPC_OUT = ROOT / "apps" / "shared" / "src" / "gateway-contract.openrpc.json"
|
|
_ENUM_PAIRS = (
|
|
("ConnectionTargetState", ConnectionTargetState, "TargetState", TargetState),
|
|
("ConnectionSettleReason", ConnectionSettleReason, "SettleReason", SettleReason),
|
|
("ConnectorAccountStatus", ConnectorAccountStatus, "ConnectionStatus", ConnectionStatus),
|
|
("ConnectorToolFacet", ConnectorToolFacet, "ConnectorTool.facet", ConnectorTool.model_fields["facet"].annotation),
|
|
("ConnectorToolsSource", ConnectorToolsSource, "ToolsRead.source", get_type_hints(ToolsRead)["source"]),
|
|
)
|
|
|
|
HEADER = (
|
|
"// GENERATED by scripts/gen_gateway_contracts.py from tui_gateway/contracts — DO NOT EDIT.\n"
|
|
"// Regenerate: .venv/bin/python scripts/gen_gateway_contracts.py\n"
|
|
"// tests/tui_gateway/contracts/test_generated.py fails when this file is stale.\n"
|
|
)
|
|
|
|
|
|
class _Schema(GenerateJsonSchema):
|
|
"""Stable ``$defs`` naming: the model's class name (no module qualifiers)."""
|
|
|
|
def normalize_name(self, name: str) -> str:
|
|
return re.sub(r"[^A-Za-z0-9_]", "_", name)
|
|
|
|
|
|
def _schema_for(models: list[type]) -> tuple[dict[str, dict], list[dict]]:
|
|
"""One shared ``$defs`` for every model, plus each model's own schema (a ``$ref`` in practice)."""
|
|
from pydantic.json_schema import models_json_schema
|
|
|
|
defs, top = models_json_schema(
|
|
[(m, "serialization") for m in models], schema_generator=_Schema, ref_template="#/$defs/{model}"
|
|
)
|
|
return top.get("$defs", {}), [defs[(m, "serialization")] for m in models]
|
|
|
|
|
|
# ── TypeScript rendering ─────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class Renderer:
|
|
def __init__(self, defs: dict[str, dict]):
|
|
self.defs = defs
|
|
self.emitted: OrderedDict[str, str] = OrderedDict()
|
|
|
|
def ref_name(self, ref: str) -> str:
|
|
assert ref.startswith("#/$defs/"), ref
|
|
return ref[len("#/$defs/"):]
|
|
|
|
def type_of(self, schema: dict, *, inline_depth: int = 0) -> str:
|
|
if "$ref" in schema:
|
|
name = self.ref_name(schema["$ref"])
|
|
self.ensure(name)
|
|
return name
|
|
if "const" in schema:
|
|
return _lit(schema["const"])
|
|
if "enum" in schema:
|
|
return " | ".join(_lit(v) for v in schema["enum"])
|
|
if "anyOf" in schema or "oneOf" in schema:
|
|
variants = schema.get("anyOf") or schema.get("oneOf") or []
|
|
rendered = list(dict.fromkeys(self.type_of(v, inline_depth=inline_depth) for v in variants))
|
|
return " | ".join(rendered)
|
|
t = schema.get("type")
|
|
if isinstance(t, list):
|
|
return " | ".join(self.type_of({**schema, "type": x}, inline_depth=inline_depth) for x in t)
|
|
if t == "string":
|
|
return "string"
|
|
if t in ("integer", "number"):
|
|
return "number"
|
|
if t == "boolean":
|
|
return "boolean"
|
|
if t == "null":
|
|
return "null"
|
|
if t == "array":
|
|
items = schema.get("items")
|
|
if items is None:
|
|
return "unknown[]"
|
|
if "prefixItems" in schema:
|
|
return "[" + ", ".join(self.type_of(x) for x in schema["prefixItems"]) + "]"
|
|
inner = self.type_of(items, inline_depth=inline_depth)
|
|
return f"({inner})[]" if " | " in inner else f"{inner}[]"
|
|
if t == "object" or "properties" in schema or "additionalProperties" in schema:
|
|
return self.object_literal(schema, inline_depth)
|
|
if not schema or set(schema) <= {"title", "description", "default"}:
|
|
return "unknown"
|
|
raise ValueError(f"unsupported JSON-Schema construct: {json.dumps(schema)[:200]}")
|
|
|
|
def object_literal(self, schema: dict, depth: int) -> str:
|
|
props = schema.get("properties")
|
|
extra = schema.get("additionalProperties")
|
|
if not props:
|
|
if extra is False:
|
|
return "Record<string, never>"
|
|
if extra in (None, True):
|
|
return "Record<string, unknown>"
|
|
return f"Record<string, {self.type_of(extra, inline_depth=depth + 1)}>"
|
|
required = set(schema.get("required", ()))
|
|
lines = ["{"]
|
|
for key, sub in props.items():
|
|
opt = "" if key in required else "?"
|
|
lines.append(f" {_prop(key)}{opt}: {self.type_of(sub, inline_depth=depth + 1)}")
|
|
if extra not in (None, False):
|
|
lines.append(f" [key: string]: {'unknown' if extra is True else self.type_of(extra)}")
|
|
lines.append("}")
|
|
return "\n".join(lines)
|
|
|
|
def ensure(self, name: str) -> None:
|
|
if name in self.emitted:
|
|
return
|
|
self.emitted[name] = "" # cycle guard
|
|
schema = self.defs[name]
|
|
doc = _doc(schema.get("description"))
|
|
if "enum" in schema:
|
|
body = f"export type {name} = {self.type_of({'enum': schema['enum']})}\n"
|
|
elif schema.get("properties"):
|
|
body = f"export interface {name} {self.object_literal(schema, 0)}\n"
|
|
else:
|
|
body = f"export type {name} = {self.type_of(schema)}\n"
|
|
self.emitted[name] = doc + body
|
|
|
|
|
|
_IDENT = re.compile(r"^[A-Za-z_$][A-Za-z0-9_$]*$")
|
|
|
|
|
|
def _prop(key: str) -> str:
|
|
return key if _IDENT.match(key) else _lit(key)
|
|
|
|
|
|
def _const_items(names: list[str]) -> str:
|
|
return ",\n".join(f" {_lit(n)}" for n in names) + "\n"
|
|
|
|
|
|
def _lit(value) -> str:
|
|
"""A TS literal in the repo's prettier style (single quotes) so the committed file needs no
|
|
Node-side formatting pass — the Python CI lane regenerates and diffs it."""
|
|
if isinstance(value, str):
|
|
return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
|
|
return json.dumps(value)
|
|
|
|
|
|
def _doc(text: str | None, indent: str = "") -> str:
|
|
if not text:
|
|
return ""
|
|
clean = " ".join(text.split())
|
|
return f"{indent}/** {clean} */\n"
|
|
|
|
|
|
def _pascal(name: str) -> str:
|
|
return "".join(p[:1].upper() + p[1:] for p in re.split(r"[._]", name))
|
|
|
|
|
|
def _enum_values(enum) -> set[str]:
|
|
values = getattr(enum, "__members__", None)
|
|
if values is not None:
|
|
return {member.value for member in values.values()}
|
|
if isinstance(enum, str):
|
|
return {enum}
|
|
return {value for member in get_args(enum) for value in _enum_values(member)}
|
|
|
|
|
|
def _check_enum_parity() -> None:
|
|
for contract_name, contract_enum, domain_name, domain_enum in _ENUM_PAIRS:
|
|
contract_values = _enum_values(contract_enum)
|
|
domain_values = _enum_values(domain_enum)
|
|
if contract_values != domain_values:
|
|
raise ValueError(
|
|
f"enum parity failed: {contract_name} vs {domain_name}: "
|
|
f"{sorted(contract_values ^ domain_values)}"
|
|
)
|
|
|
|
|
|
def render_ts() -> str:
|
|
_check_enum_parity()
|
|
models: list[type] = []
|
|
for m in METHODS.values():
|
|
models += [m.params, m.result]
|
|
for r in SERVER_REQUESTS.values():
|
|
models += [r.params, r.result]
|
|
for e in EVENTS.values():
|
|
if e.payload is not None:
|
|
models.append(e.payload)
|
|
# de-dup preserving order
|
|
seen: dict[type, None] = OrderedDict()
|
|
for m in models:
|
|
seen.setdefault(m)
|
|
models = list(seen)
|
|
defs, tops = _schema_for(models)
|
|
r = Renderer(defs)
|
|
name_of = {m: r.ref_name(t["$ref"]) for m, t in zip(models, tops)}
|
|
for m in models:
|
|
r.ensure(name_of[m])
|
|
|
|
out = [HEADER, "/* eslint-disable */\n", "// ── Types ──\n"]
|
|
out.extend(r.emitted.values())
|
|
out.append(
|
|
"export type ConnectorErrorReason = "
|
|
+ r.type_of({"enum": [member.value for member in ConnectorErrorReason]})
|
|
+ "\n"
|
|
)
|
|
|
|
out.append("\n// ── Client→server methods ──\n")
|
|
out.append("export interface RpcMethods {\n")
|
|
for m in sorted(METHODS.values(), key=lambda x: x.name):
|
|
out.append(_doc(m.doc, " "))
|
|
out.append(f" {_prop(m.name)}: {{ params: {name_of[m.params]}; result: {name_of[m.result]} }}\n")
|
|
out.append("}\n")
|
|
out.append("export type RpcMethod = keyof RpcMethods\n")
|
|
out.append("export const RPC_METHODS = [\n" + _const_items(sorted(METHODS)) + "] as const satisfies readonly RpcMethod[]\n")
|
|
|
|
out.append("\n// ── Server→client requests ──\n")
|
|
out.append("export interface ServerRequestMap {\n")
|
|
for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name):
|
|
out.append(_doc(s.doc, " "))
|
|
out.append(f" {_prop(s.name)}: {{ params: {name_of[s.params]}; result: {name_of[s.result]} }}\n")
|
|
out.append("}\n")
|
|
out.append("export type ServerRequestMethod = keyof ServerRequestMap\n")
|
|
out.append("export const SERVER_REQUEST_METHODS = [\n" + _const_items(sorted(SERVER_REQUESTS))
|
|
+ "] as const satisfies readonly ServerRequestMethod[]\n")
|
|
|
|
out.append("\n// ── Notifications (`event` frames) ──\n")
|
|
out.append("export interface BackendGatewayEventMap {\n")
|
|
for e in sorted(EVENTS.values(), key=lambda x: x.name):
|
|
out.append(_doc(e.doc, " "))
|
|
payload = name_of[e.payload] if e.payload is not None else "Record<string, never>"
|
|
out.append(f" {_prop(e.name)}: {payload}\n")
|
|
out.append("}\n")
|
|
out.append("export type BackendGatewayEventName = keyof BackendGatewayEventMap\n")
|
|
out.append("export const GATEWAY_EVENT_TYPES = [\n" + _const_items(sorted(EVENTS))
|
|
+ "] as const satisfies readonly BackendGatewayEventName[]\n")
|
|
return "".join(out)
|
|
|
|
|
|
def _tidy(text: str) -> str:
|
|
"""No trailing whitespace, single trailing newline (matches `git diff --check` + prettier)."""
|
|
return "\n".join(line.rstrip() for line in text.splitlines()).rstrip("\n") + "\n"
|
|
|
|
|
|
# ── OpenRPC rendering ────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _openrpc_schema(model: type) -> dict:
|
|
schema = TypeAdapter(model).json_schema(schema_generator=_Schema, ref_template="#/components/schemas/{model}")
|
|
schema.pop("$defs", None)
|
|
return schema
|
|
|
|
|
|
def render_openrpc() -> str:
|
|
_check_enum_parity()
|
|
components: dict[str, dict] = {}
|
|
all_models: list[type] = []
|
|
for m in METHODS.values():
|
|
all_models += [m.params, m.result]
|
|
for r in SERVER_REQUESTS.values():
|
|
all_models += [r.params, r.result]
|
|
for e in EVENTS.values():
|
|
if e.payload is not None:
|
|
all_models.append(e.payload)
|
|
from pydantic.json_schema import models_json_schema
|
|
|
|
seen: dict[type, None] = OrderedDict()
|
|
for m in all_models:
|
|
seen.setdefault(m)
|
|
_, top = models_json_schema(
|
|
[(m, "serialization") for m in seen], schema_generator=_Schema,
|
|
ref_template="#/components/schemas/{model}",
|
|
)
|
|
components = top.get("$defs", {})
|
|
components["ConnectorErrorReason"] = {
|
|
"type": "string",
|
|
"enum": [member.value for member in ConnectorErrorReason],
|
|
}
|
|
|
|
def ref(model: type) -> dict:
|
|
return {"$ref": f"#/components/schemas/{model.__name__}"}
|
|
|
|
doc = {
|
|
"openrpc": "1.3.2",
|
|
"info": {"title": "Hermes TUI/Desktop gateway", "version": "1",
|
|
"description": "Generated from tui_gateway/contracts by scripts/gen_gateway_contracts.py."},
|
|
"methods": [
|
|
{"name": m.name, "summary": " ".join(m.doc.split()),
|
|
"params": [{"name": "params", "schema": ref(m.params)}],
|
|
"result": {"name": "result", "schema": ref(m.result)}}
|
|
for m in sorted(METHODS.values(), key=lambda x: x.name)
|
|
],
|
|
"components": {"schemas": components},
|
|
"x-server-requests": [
|
|
{"name": s.name, "summary": " ".join(s.doc.split()),
|
|
"params": [{"name": "params", "schema": ref(s.params)}],
|
|
"result": {"name": "result", "schema": ref(s.result)}}
|
|
for s in sorted(SERVER_REQUESTS.values(), key=lambda x: x.name)
|
|
],
|
|
"x-notifications": [
|
|
{"name": e.name, "summary": " ".join(e.doc.split()),
|
|
"params": [{"name": "payload", "schema": ref(e.payload) if e.payload is not None
|
|
else {"type": "object", "additionalProperties": False}}]}
|
|
for e in sorted(EVENTS.values(), key=lambda x: x.name)
|
|
],
|
|
}
|
|
return json.dumps(doc, indent=2, sort_keys=False) + "\n"
|
|
|
|
|
|
def render_all() -> dict[Path, str]:
|
|
return {TS_OUT: _tidy(render_ts()), OPENRPC_OUT: render_openrpc()}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = argv if argv is not None else sys.argv[1:]
|
|
check = "--check" in args
|
|
stale = []
|
|
for path, text in render_all().items():
|
|
current = path.read_text(encoding="utf-8-sig") if path.exists() else None
|
|
if current == text:
|
|
continue
|
|
if check:
|
|
stale.append(path)
|
|
else:
|
|
path.write_text(text, encoding="utf-8")
|
|
print(f"wrote {path.relative_to(ROOT)}")
|
|
if stale:
|
|
for p in stale:
|
|
print(f"stale: {p.relative_to(ROOT)} — run scripts/gen_gateway_contracts.py", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|