pulse/scripts/release_control/generate_mobile_compatibility.py

#!/usr/bin/env python3
"""Generate runtime and mobile projections of the Pulse Mobile contract."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parents[2]
MANIFEST_REL = Path("docs/release-control/v6/internal/MOBILE_COMPATIBILITY_MANIFEST.json")
API_OUTPUT_REL = Path("internal/api/relay_mobile_capability_generated.go")
RELAY_OUTPUT_REL = Path("internal/relay/mobile_compatibility_generated.go")
MOBILE_OUTPUT_REL = Path("src/generated/coreCompatibility.ts")

METHOD_SYMBOLS = {
    "DELETE": "http.MethodDelete",
    "GET": "http.MethodGet",
    "PATCH": "http.MethodPatch",
    "POST": "http.MethodPost",
    "PUT": "http.MethodPut",
}

SCOPE_SYMBOLS = {
    "actions:approve": "config.ScopeActionsApprove",
    "actions:execute": "config.ScopeActionsExecute",
    "ai:chat": "config.ScopeAIChat",
    "ai:execute": "config.ScopeAIExecute",
    "monitoring:read": "config.ScopeMonitoringRead",
    "monitoring:write": "config.ScopeMonitoringWrite",
    "settings:read": "config.ScopeSettingsRead",
}

VALID_FIELD_TYPES = {"array", "boolean", "integer", "number", "object", "string"}
VALID_LIFECYCLES = {"compatibility", "current"}
VALID_IMPACTS = {"desktop-only", "mobile-handoff-only", "mobile-required"}
PLACEHOLDER_RE = re.compile(r"\{([a-z][a-z0-9_]*)\}")
ROUTE_ID_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
GO_SYMBOL_RE = re.compile(r"^[A-Za-z][A-Za-z0-9]*$")


class ManifestError(ValueError):
    """Raised when the canonical manifest is internally inconsistent."""


def load_manifest(pulse_root: Path) -> tuple[dict[str, Any], bytes]:
    path = pulse_root / MANIFEST_REL
    raw = path.read_bytes()
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ManifestError(f"invalid JSON in {path}: {exc}") from exc
    if not isinstance(payload, dict):
        raise ManifestError("mobile compatibility manifest root must be an object")
    validate_manifest(payload)
    return payload, raw


def _require_non_empty_string(value: Any, label: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ManifestError(f"{label} must be a non-empty string")
    return value


def _reject_unknown_keys(
    value: dict[str, Any], allowed: set[str], label: str
) -> None:
    unknown = set(value) - allowed
    if unknown:
        raise ManifestError(f"{label} has unknown keys: {sorted(unknown)}")


def _validate_symbol_values(
    values: Any, label: str, *, extra_keys: set[str] | None = None
) -> None:
    if not isinstance(values, list) or not values:
        raise ManifestError(f"{label} must be a non-empty list")
    seen_values: set[str] = set()
    seen_symbols: set[str] = set()
    for index, entry in enumerate(values):
        if not isinstance(entry, dict):
            raise ManifestError(f"{label}[{index}] must be an object")
        _reject_unknown_keys(
            entry,
            {"value", "go_symbol"} | (extra_keys or set()),
            f"{label}[{index}]",
        )
        value = _require_non_empty_string(entry.get("value"), f"{label}[{index}].value")
        symbol = _require_non_empty_string(
            entry.get("go_symbol"), f"{label}[{index}].go_symbol"
        )
        if not GO_SYMBOL_RE.fullmatch(symbol):
            raise ManifestError(f"{label}[{index}].go_symbol is invalid")
        if value in seen_values:
            raise ManifestError(f"duplicate {label} value {value!r}")
        if symbol in seen_symbols:
            raise ManifestError(f"duplicate {label} Go symbol {symbol!r}")
        seen_values.add(value)
        seen_symbols.add(symbol)


def _validate_fields(fields: Any, label: str) -> None:
    if not isinstance(fields, list):
        raise ManifestError(f"{label} must be a list")
    seen: set[str] = set()
    for index, field in enumerate(fields):
        if not isinstance(field, dict):
            raise ManifestError(f"{label}[{index}] must be an object")
        _reject_unknown_keys(field, {"path", "type", "required"}, f"{label}[{index}]")
        path = _require_non_empty_string(field.get("path"), f"{label}[{index}].path")
        field_type = _require_non_empty_string(
            field.get("type"), f"{label}[{index}].type"
        )
        if field_type not in VALID_FIELD_TYPES:
            raise ManifestError(f"{label}[{index}] has unknown type {field_type!r}")
        if "required" in field and not isinstance(field["required"], bool):
            raise ManifestError(f"{label}[{index}].required must be boolean")
        if path in seen:
            raise ManifestError(f"duplicate field {path!r} in {label}")
        seen.add(path)


def validate_manifest(payload: dict[str, Any]) -> None:
    _reject_unknown_keys(
        payload,
        {
            "$schema",
            "schema_version",
            "contract_version",
            "mobile_role",
            "surface_classes",
            "pairing",
            "push",
            "routes",
        },
        "manifest",
    )
    if payload.get("schema_version") != 1:
        raise ManifestError("schema_version must be 1")
    if not isinstance(payload.get("contract_version"), int) or payload["contract_version"] < 1:
        raise ManifestError("contract_version must be a positive integer")
    _require_non_empty_string(payload.get("mobile_role"), "mobile_role")

    classes = payload.get("surface_classes")
    if not isinstance(classes, list):
        raise ManifestError("surface_classes must be a list")
    for index, entry in enumerate(classes):
        if not isinstance(entry, dict):
            raise ManifestError(f"surface_classes[{index}] must be an object")
        _reject_unknown_keys(
            entry, {"id", "description"}, f"surface_classes[{index}]"
        )
        _require_non_empty_string(
            entry.get("description"), f"surface_classes[{index}].description"
        )
    class_ids = [entry.get("id") for entry in classes if isinstance(entry, dict)]
    if set(class_ids) != VALID_IMPACTS or len(class_ids) != len(VALID_IMPACTS):
        raise ManifestError(
            "surface_classes must define mobile-required, mobile-handoff-only, and desktop-only exactly once"
        )

    pairing = payload.get("pairing")
    if not isinstance(pairing, dict):
        raise ManifestError("pairing must be an object")
    _reject_unknown_keys(
        pairing,
        {"schema", "scheme", "host", "required_fields", "optional_fields"},
        "pairing",
    )
    for key in ("schema", "scheme", "host"):
        _require_non_empty_string(pairing.get(key), f"pairing.{key}")
    required_pairing = pairing.get("required_fields")
    optional_pairing = pairing.get("optional_fields")
    if not isinstance(required_pairing, list) or not all(
        isinstance(value, str) and value for value in required_pairing
    ):
        raise ManifestError("pairing.required_fields must be a string list")
    if not isinstance(optional_pairing, list) or not all(
        isinstance(value, str) and value for value in optional_pairing
    ):
        raise ManifestError("pairing.optional_fields must be a string list")
    if len(required_pairing) != len(set(required_pairing)):
        raise ManifestError("pairing.required_fields must not contain duplicates")
    if len(optional_pairing) != len(set(optional_pairing)):
        raise ManifestError("pairing.optional_fields must not contain duplicates")
    overlap = set(required_pairing) & set(optional_pairing)
    if overlap:
        raise ManifestError(f"pairing fields cannot be both required and optional: {sorted(overlap)}")

    push = payload.get("push")
    if not isinstance(push, dict):
        raise ManifestError("push must be an object")
    _reject_unknown_keys(
        push, {"required_fields", "types", "priorities", "actions"}, "push"
    )
    required_push_fields = push.get("required_fields")
    if not isinstance(required_push_fields, list) or not all(
        isinstance(value, str) and value for value in required_push_fields
    ):
        raise ManifestError("push.required_fields must be a string list")
    if len(required_push_fields) != len(set(required_push_fields)):
        raise ManifestError("push.required_fields must not contain duplicates")
    _validate_symbol_values(push.get("types"), "push.types")
    _validate_symbol_values(push.get("priorities"), "push.priorities")
    _validate_symbol_values(
        push.get("actions"),
        "push.actions",
        extra_keys={"destination", "lifecycle"},
    )
    for index, action in enumerate(push["actions"]):
        if action.get("lifecycle") not in VALID_LIFECYCLES:
            raise ManifestError(f"push.actions[{index}] has invalid lifecycle")
        if action.get("destination") not in {"approval-detail", "finding-detail"}:
            raise ManifestError(f"push.actions[{index}] has invalid destination")

    routes = payload.get("routes")
    if not isinstance(routes, list) or not routes:
        raise ManifestError("routes must be a non-empty list")
    seen_ids: set[str] = set()
    seen_symbols: set[str] = set()
    seen_method_paths: set[tuple[str, str]] = set()
    for index, route in enumerate(routes):
        if not isinstance(route, dict):
            raise ManifestError(f"routes[{index}] must be an object")
        _reject_unknown_keys(
            route,
            {
                "id",
                "go_symbol",
                "method",
                "path",
                "required_scope",
                "legacy_scope",
                "lifecycle",
                "mobile_impact",
                "implementation_paths",
                "request_fields",
                "response_fields",
            },
            f"routes[{index}]",
        )
        route_id = _require_non_empty_string(route.get("id"), f"routes[{index}].id")
        symbol = _require_non_empty_string(
            route.get("go_symbol"), f"routes[{index}].go_symbol"
        )
        method = _require_non_empty_string(route.get("method"), f"routes[{index}].method")
        path = _require_non_empty_string(route.get("path"), f"routes[{index}].path")
        scope = _require_non_empty_string(
            route.get("required_scope"), f"routes[{index}].required_scope"
        )
        if not ROUTE_ID_RE.fullmatch(route_id):
            raise ManifestError(f"route id {route_id!r} is invalid")
        if not GO_SYMBOL_RE.fullmatch(symbol):
            raise ManifestError(f"route {route_id!r} Go symbol is invalid")
        if method not in METHOD_SYMBOLS:
            raise ManifestError(f"route {route_id!r} has unsupported method {method!r}")
        if scope not in SCOPE_SYMBOLS:
            raise ManifestError(f"route {route_id!r} has unsupported scope {scope!r}")
        legacy_scope = route.get("legacy_scope")
        if legacy_scope is not None and legacy_scope not in SCOPE_SYMBOLS:
            raise ManifestError(
                f"route {route_id!r} has unsupported legacy scope {legacy_scope!r}"
            )
        if route.get("lifecycle") not in VALID_LIFECYCLES:
            raise ManifestError(f"route {route_id!r} has invalid lifecycle")
        if route.get("mobile_impact") not in VALID_IMPACTS:
            raise ManifestError(f"route {route_id!r} has invalid mobile_impact")
        implementation_paths = route.get("implementation_paths")
        if (
            not isinstance(implementation_paths, list)
            or not implementation_paths
            or not all(
                isinstance(implementation_path, str) and implementation_path
                for implementation_path in implementation_paths
            )
        ):
            raise ManifestError(f"route {route_id!r} must name implementation_paths")
        if len(implementation_paths) != len(set(implementation_paths)):
            raise ManifestError(f"route {route_id!r} has duplicate implementation_paths")
        _validate_fields(route.get("request_fields"), f"route {route_id} request_fields")
        _validate_fields(route.get("response_fields"), f"route {route_id} response_fields")
        if route_id in seen_ids:
            raise ManifestError(f"duplicate route id {route_id!r}")
        if symbol in seen_symbols:
            raise ManifestError(f"duplicate route Go symbol {symbol!r}")
        if (method, path) in seen_method_paths:
            raise ManifestError(f"duplicate route {method} {path}")
        if not path.startswith("/api/"):
            raise ManifestError(f"route {route_id!r} path must start with /api/")
        placeholders = PLACEHOLDER_RE.findall(path)
        if len(placeholders) != len(set(placeholders)):
            raise ManifestError(f"route {route_id!r} repeats a path placeholder")
        seen_ids.add(route_id)
        seen_symbols.add(symbol)
        seen_method_paths.add((method, path))


def manifest_sha256(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def render_api_go(payload: dict[str, Any], digest: str) -> str:
    routes = payload["routes"]
    lines = [
        "// Code generated by scripts/release_control/generate_mobile_compatibility.py. DO NOT EDIT.",
        f"// Source SHA256: {digest}",
        "",
        "package api",
        "",
        "import (",
        '\t"net/http"',
        "",
        '\t"github.com/rcourtman/pulse-go-rewrite/internal/config"',
        ")",
        "",
        f'const onboardingSchemaVersion = {json.dumps(payload["pairing"]["schema"])}',
        "",
        "const (",
    ]
    for route in routes:
        lines.append(
            f'\t{route["go_symbol"]} relayMobileRuntimeRouteID = {json.dumps(route["id"])}'
        )
    lines.extend([")", "", "var relayMobileRuntimeRouteOrder = []relayMobileRuntimeRouteID{"])
    lines.extend(f'\t{route["go_symbol"]},' for route in routes)
    lines.extend(["}", "", "var relayMobileRuntimeRouteSpecs = map[relayMobileRuntimeRouteID]relayMobileRuntimeRouteSpec{"])
    for route in routes:
        lines.extend(
            [
                f'\t{route["go_symbol"]}: {{',
                f'\t\tid: {route["go_symbol"]},',
                f'\t\tmethod: {METHOD_SYMBOLS[route["method"]]},',
                f'\t\tpath: {json.dumps(route["path"])},',
                f'\t\trequiredScope: {SCOPE_SYMBOLS[route["required_scope"]]},',
            ]
        )
        if route.get("legacy_scope"):
            lines.append(f'\t\tlegacyScope: {SCOPE_SYMBOLS[route["legacy_scope"]]},')
        lines.append("\t},")
    lines.extend(["}", ""])
    return "\n".join(lines)


def render_relay_go(payload: dict[str, Any], digest: str) -> str:
    push = payload["push"]
    lines = [
        "// Code generated by scripts/release_control/generate_mobile_compatibility.py. DO NOT EDIT.",
        f"// Source SHA256: {digest}",
        "",
        "package relay",
        "",
        "// Notification type constants consumed by Pulse Mobile.",
        "const (",
    ]
    lines.extend(f'\t{entry["go_symbol"]} = {json.dumps(entry["value"])}' for entry in push["types"])
    lines.extend([")", "", "// Priority constants consumed by Pulse Mobile.", "const ("])
    lines.extend(
        f'\t{entry["go_symbol"]} = {json.dumps(entry["value"])}'
        for entry in push["priorities"]
    )
    lines.extend([")", "", "// Action type constants consumed by Pulse Mobile.", "const ("])
    lines.extend(
        f'\t{entry["go_symbol"]} = {json.dumps(entry["value"])}' for entry in push["actions"]
    )
    lines.extend([")", ""])
    return "\n".join(lines)


def render_mobile_typescript(payload: dict[str, Any], digest: str) -> str:
    routes = {
        route["id"]: {
            "method": route["method"],
            "path": route["path"],
            "requiredScope": route["required_scope"],
            "compatibleScopes": list(
                dict.fromkeys(
                    [
                        "relay:mobile:access",
                        route["required_scope"],
                        *([route["legacy_scope"]] if route.get("legacy_scope") else []),
                    ]
                )
            ),
            "lifecycle": route["lifecycle"],
            "mobileImpact": route["mobile_impact"],
        }
        for route in payload["routes"]
    }
    push_actions = {
        action["value"]: {
            "value": action["value"],
            "destination": action["destination"],
            "lifecycle": action["lifecycle"],
        }
        for action in payload["push"]["actions"]
    }
    return "\n".join(
        [
            "// This file is generated by Pulse's mobile compatibility generator.",
            "// Do not edit it in pulse-mobile; update MOBILE_COMPATIBILITY_MANIFEST.json in Pulse.",
            f"// Source SHA256: {digest}",
            "",
            f"export const CORE_MOBILE_CONTRACT_VERSION = {payload['contract_version']} as const;",
            f"export const CORE_MOBILE_CONTRACT_SHA256 = {json.dumps(digest)} as const;",
            "",
            "export const CORE_MOBILE_PAIRING = "
            + json.dumps(payload["pairing"], indent=2, ensure_ascii=False)
            + " as const;",
            "",
            "export const CORE_MOBILE_ROUTES = "
            + json.dumps(routes, indent=2, ensure_ascii=False)
            + " as const;",
            "",
            "export type CoreMobileRouteId = keyof typeof CORE_MOBILE_ROUTES;",
            "",
            "export const CORE_MOBILE_PUSH = "
            + json.dumps(
                {
                    "requiredFields": payload["push"]["required_fields"],
                    "types": [entry["value"] for entry in payload["push"]["types"]],
                    "priorities": [
                        entry["value"] for entry in payload["push"]["priorities"]
                    ],
                    "actions": push_actions,
                },
                indent=2,
                ensure_ascii=False,
            )
            + " as const;",
            "",
            "export type CoreMobilePushAction = keyof typeof CORE_MOBILE_PUSH.actions;",
            "",
        ]
    )


def format_go(source: str) -> str:
    completed = subprocess.run(
        ["gofmt"],
        input=source,
        check=True,
        capture_output=True,
        text=True,
    )
    return completed.stdout


def generated_outputs(
    pulse_root: Path, mobile_repo: Path | None
) -> dict[Path, str]:
    payload, raw = load_manifest(pulse_root)
    digest = manifest_sha256(raw)
    outputs = {
        pulse_root / API_OUTPUT_REL: format_go(render_api_go(payload, digest)),
        pulse_root / RELAY_OUTPUT_REL: format_go(render_relay_go(payload, digest)),
    }
    if mobile_repo is not None:
        outputs[mobile_repo / MOBILE_OUTPUT_REL] = render_mobile_typescript(payload, digest)
    return outputs


def write_outputs(outputs: dict[Path, str]) -> None:
    for path, content in outputs.items():
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")


def check_outputs(outputs: dict[Path, str]) -> list[str]:
    errors: list[str] = []
    for path, expected in outputs.items():
        try:
            actual = path.read_text(encoding="utf-8")
        except OSError:
            errors.append(f"missing generated projection: {path}")
            continue
        if actual != expected:
            errors.append(f"generated projection is stale: {path}")
    return errors


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--pulse-root", default=str(REPO_ROOT))
    parser.add_argument("--mobile-repo", default="")
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--write", action="store_true")
    mode.add_argument("--check", action="store_true")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    pulse_root = Path(args.pulse_root).resolve()
    mobile_repo = Path(args.mobile_repo).resolve() if args.mobile_repo else None
    try:
        outputs = generated_outputs(pulse_root, mobile_repo)
        if args.write:
            write_outputs(outputs)
            print(f"[OK] Wrote {len(outputs)} mobile compatibility projections")
            return 0
        errors = check_outputs(outputs)
    except (ManifestError, OSError, subprocess.CalledProcessError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    if errors:
        for error in errors:
            print(f"ERROR: {error}", file=sys.stderr)
        print(
            "Run scripts/release_control/generate_mobile_compatibility.py --write "
            "--mobile-repo ../pulse-mobile",
            file=sys.stderr,
        )
        return 1
    print(f"[OK] {len(outputs)} mobile compatibility projections are current")
    return 0


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