open-standards/scripts/check-wai-ops-doors.py

#!/usr/bin/env python3
"""check-wai-ops-doors.py: every door of WAI names its operations as the catalogue does.

WAI's operations are one catalogue, `noun.verb` (wai/wai-rs/src/catalogue.rs), exported as
wai/catalogue/ops.json by `wai ops export` (CI holds the export to the table with
`wai ops export --check`). Each door names an operation by a mechanical transform of its id:

  MCP          tool `noun_verb` (or its `op`)        wai/wai-rs/src/bin/wai_cli/mcp/tools.rs
  HTTP API     operationId `noun.verb`              every openapi.json of the services
  C            function `wai_noun_verb`             wai/wai-rs/include/wai.h
  Swift        type `Wai<Noun>`, method `<verb>…`   wai/wai-sink-swift/Sources
  Kotlin       class `Wai<Noun>`, fun `<verb>…`     wai/wai-sink-android/*/src/main

The command line is held by the `wai` binary's own tests (wai_cli/door.rs).

What each door is held to:

- MCP: every tool's name is `noun_verb` of an operation in the catalogue.
- HTTP API: every operationId is an operation of the catalogue that runs on a service
  (placement `service` or `both`), at the route the catalogue gives it.
- C, Swift, Kotlin: a name in a catalogue noun's namespace whose verb is an operation verb
  (the closed list below) names an operation, and that operation must be in the catalogue;
  a near-synonym verb (`get`, `describe`, `info`, `keygen`, `init`, `new`) in a catalogue
  noun's namespace is refused. Any other name is library API the catalogue does not list
  (the C ABI's codec and sink functions, the native sinks' classes), and is reported as such.

A door that has no operation of the catalogue yet is reported as such; that is not a
failure. A wrong name is.

    check-wai-ops-doors.py [--root DIR] [--ops FILE]

Exit 0 when every door holds, 1 with one line a finding otherwise. Standard library only.
"""
import glob
import json
import os
import re
import sys

# The operation verbs: the closed list of WAI toolchain v1 (naming rule N3).
VERBS = {
    "create", "list", "show", "update", "delete", "check", "verify", "run", "build", "export",
    "import", "start", "stop", "join", "leave", "grant", "invite", "revoke", "rotate", "tail",
    "search", "test", "load", "encode", "decode", "inspect", "wrap", "unwrap", "fetch",
    "generate", "promote", "rollback", "lock", "upgrade", "setup", "guide", "stream",
}
# Near-synonyms of an operation verb, and the verb to use.
NEAR = {"get": "show", "describe": "show", "info": "show", "keygen": "create", "init": "create", "new": "create"}

MCP_TABLE = "wai/wai-rs/src/bin/wai_cli/mcp/tools.rs"
OPENAPI_GLOBS = ["joule-code/joule-code-rs/crates/*/openapi.json", "wai/*/openapi.json", "wai/wai-rs/openapi.json"]
C_HEADER = "wai/wai-rs/include/wai.h"
SWIFT_GLOB = "wai/wai-sink-swift/Sources/**/*.swift"
KOTLIN_GLOB = "wai/wai-sink-android/*/src/main/**/*.kt"


def load_ops(path):
    with open(path, encoding="utf-8") as f:
        doc = json.load(f)
    if doc.get("format") != "wai.ops/1":
        raise SystemExit(f"{path}: not a catalogue export (format {doc.get('format')!r})")
    return {o["id"]: o for o in doc["operations"]}


def split_id(i):
    n, _, v = i.partition(".")
    return n, v


def check_mcp(root, ops, findings, notes):
    path = os.path.join(root, MCP_TABLE)
    if not os.path.exists(path):
        notes.append("MCP: no MCP door yet (no tool table); nothing to hold")
        return
    with open(path, encoding="utf-8") as f:
        src = f.read()
    # A tool is named outright (`name: "noun_verb"`) or by its operation (`op: "noun.verb"`),
    # its name then the operation's with `_` for the dot.
    names = []
    for body in re.findall(r"\bTool\s*\{(.*?)\}\s*,", src, re.S):
        named = re.search(r'\bname:\s*"([^"]+)"', body)
        by_op = re.search(r'\bop:\s*"([^"]+)"', body)
        if named:
            names.append(named.group(1))
        elif by_op:
            names.append(by_op.group(1).replace(".", "_", 1))
    if not names:
        findings.append(f"MCP: {MCP_TABLE}: no tool name could be read (`Tool {{ name: \"…\" }}` or `Tool {{ op: \"…\" }}`)")
        return
    bad = 0
    for t in names:
        if t.count("_") != 1:
            findings.append(f"MCP: tool {t!r} is not noun_verb of an operation")
            bad += 1
            continue
        i = t.replace("_", ".")
        if i not in ops:
            findings.append(f"MCP: tool {t!r} names {i}, which the catalogue does not hold")
            bad += 1
    if not bad:
        notes.append(f"MCP: {len(names)} tools, each an operation of the catalogue")


def openapi_operations(doc):
    for path, item in (doc.get("paths") or {}).items():
        for method, op in item.items():
            if method.lower() in ("get", "put", "post", "delete", "patch", "head", "options", "trace") and isinstance(op, dict):
                yield method.upper(), path, op.get("operationId")


def check_openapi(root, ops, findings, notes):
    files = sorted({f for g in OPENAPI_GLOBS for f in glob.glob(os.path.join(root, g))})
    if not files:
        notes.append("HTTP API: no OpenAPI document yet; nothing to hold")
        return
    for f in files:
        rel = os.path.relpath(f, root)
        with open(f, encoding="utf-8") as fh:
            doc = json.load(fh)
        n = bad = 0
        for method, path, oid in openapi_operations(doc):
            n += 1
            where = f"HTTP API: {rel}: {method} {path}"
            if not oid:
                findings.append(f"{where}: no operationId")
                bad += 1
            elif oid not in ops:
                findings.append(f"{where}: operationId {oid!r} is not an operation of the catalogue")
                bad += 1
            else:
                o = ops[oid]
                if o["placement"] not in ("service", "both"):
                    findings.append(f"{where}: {oid} is placed {o['placement']}, not on a service")
                    bad += 1
                r = o.get("route")
                if r and (r["method"].upper(), r["path"]) != (method, path):
                    findings.append(f"{where}: the catalogue routes {oid} to {r['method']} {r['path']}")
                    bad += 1
        if not bad:
            notes.append(f"HTTP API: {rel}: {n} operations, each an operation of the catalogue at its route")


def held_name(door, where, noun, verb, nouns, ops, findings):
    """A name in `noun`'s namespace with `verb`: a finding when it names an operation the
    catalogue does not hold, or uses a near-synonym. True when it names an operation."""
    if noun not in nouns:
        return False
    if verb in NEAR:
        findings.append(f"{door}: {where}: the verb {verb} is a near-synonym; the operation verb is {NEAR[verb]}")
        return True
    if verb not in VERBS:
        return False
    i = f"{noun}.{verb}"
    if i not in ops:
        near = sorted(o for o in ops if split_id(o)[1] == verb)
        hint = f" (operations with that verb: {', '.join(near)})" if near else ""
        findings.append(f"{door}: {where} names {i}, which the catalogue does not hold{hint}")
    return True


def check_c(root, ops, nouns, findings, notes):
    path = os.path.join(root, C_HEADER)
    if not os.path.exists(path):
        notes.append("C: no header; nothing to hold")
        return
    with open(path, encoding="utf-8") as f:
        src = f.read()
    fns = sorted(set(re.findall(r"\b(wai_[a-z0-9_]+)\s*\(", src)))
    mapped = other = 0
    before = len(findings)
    for fn in fns:
        parts = fn.split("_")[1:]
        if len(parts) >= 2 and held_name("C", fn, parts[0], parts[1], nouns, ops, findings):
            mapped += 1
        else:
            other += 1
    if len(findings) == before:
        if mapped:
            notes.append(f"C: {mapped} functions name operations of the catalogue, each held; {other} are library API")
        else:
            notes.append(f"C: no operation of the catalogue in this door yet; its {other} functions are library API (codecs, envelopes, the sink)")


def camel_verb(name):
    m = re.match(r"[a-z]+", name)
    return m.group(0) if m else ""


TYPE_RE = {
    "Swift": re.compile(r"^\s*(?:@\w+\s+)*(public|open)\s+(?:final\s+)?(?:class|struct|enum|actor)\s+(\w+)|^\s*(?:public\s+)?extension\s+(\w+)"),
    "Kotlin": re.compile(r"^\s*(?!private|internal)(?:public\s+)?(?:open\s+|abstract\s+|sealed\s+|data\s+|enum\s+|final\s+)*(?:class|object|interface)\s+(\w+)"),
}
FUNC_RE = {
    "Swift": re.compile(r"^\s*(?:@\w+\s+)*(?:public|open)\s+(?:static\s+|class\s+|final\s+|override\s+)*func\s+(\w+)"),
    "Kotlin": re.compile(r"^\s*(?!.*\b(?:private|internal|protected)\b)(?:public\s+|override\s+|suspend\s+|inline\s+|operator\s+)*fun\s+(?:<[^>]*>\s*)?(\w+)"),
}


def sdk_names(lang, src):
    """Every (type, method) the source declares public: a method under the type whose body
    holds it, a companion object's under its class. A type's body opens at the first line
    after its declaration that ends with `{` (a Kotlin primary constructor can take several
    lines, and a default argument can hold a lambda's braces)."""
    out = []
    stack = []  # (type name, brace depth inside its body)
    depth = 0
    pending = None
    for line in src.splitlines():
        code = re.sub(r'"(?:\\.|[^"\\])*"', '""', line.split("//")[0])
        name = None
        t = TYPE_RE[lang].match(code)
        if t:
            name = next(g for g in t.groups() if g and g not in ("public", "open"))
        elif lang == "Kotlin" and re.match(r"^\s*companion\s+object\b", code) and stack:
            name = stack[-1][0]
        f = FUNC_RE[lang].match(code)
        if f and stack:
            out.append((stack[-1][0], f.group(1)))
        if name:
            pending = name
        if pending and code.rstrip().endswith("{"):
            stack.append((pending, depth + 1))
            pending = None
        for ch in code:
            if ch == "{":
                depth += 1
            elif ch == "}":
                depth -= 1
                while stack and depth < stack[-1][1]:
                    stack.pop()
    return out


def check_sdk(lang, root, pattern, ops, nouns, findings, notes):
    files = sorted(glob.glob(os.path.join(root, pattern), recursive=True))
    if not files:
        notes.append(f"{lang}: no sources; nothing to hold")
        return
    mapped = other = 0
    before = len(findings)
    for f in files:
        with open(f, encoding="utf-8") as fh:
            src = fh.read()
        for typ, meth in sdk_names(lang, src):
            if not typ.startswith("Wai") or len(typ) <= 3:
                other += 1
                continue
            noun = typ[3:].lower()
            where = f"{os.path.relpath(f, root)}: {typ}.{meth}"
            if held_name(lang, where, noun, camel_verb(meth), nouns, ops, findings):
                mapped += 1
            else:
                other += 1
    if len(findings) == before:
        if mapped:
            notes.append(f"{lang}: {mapped} methods name operations of the catalogue, each held; {other} are library API")
        else:
            notes.append(f"{lang}: no operation of the catalogue in this door yet; its {other} public names are the native sink's API, whose nouns the catalogue does not hold")


def main(argv):
    root, ops_path = ".", None
    i = 0
    while i < len(argv):
        if argv[i] == "--root" and i + 1 < len(argv):
            root = argv[i + 1]
            i += 2
        elif argv[i] == "--ops" and i + 1 < len(argv):
            ops_path = argv[i + 1]
            i += 2
        else:
            print(__doc__.strip().splitlines()[0], file=sys.stderr)
            print("usage: check-wai-ops-doors.py [--root DIR] [--ops FILE]", file=sys.stderr)
            return 2
    ops = load_ops(ops_path or os.path.join(root, "wai/catalogue/ops.json"))
    nouns = {split_id(i)[0] for i in ops}
    findings, notes = [], []
    check_mcp(root, ops, findings, notes)
    check_openapi(root, ops, findings, notes)
    check_c(root, ops, nouns, findings, notes)
    check_sdk("Swift", root, SWIFT_GLOB, ops, nouns, findings, notes)
    check_sdk("Kotlin", root, KOTLIN_GLOB, ops, nouns, findings, notes)
    for n in notes:
        print(f"ok    {n}")
    for f in findings:
        print(f"FAIL  {f}")
    print(f"check-wai-ops-doors: {len(ops)} operations; {len(findings)} finding(s)")
    return 1 if findings else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))