scripts.gen_cxp

Reference › scripts › gen_cxp

scripts/gen_cxp.py regenerates the CXP protobuf message stubs cxp_pb2.py and cxp_pb2.pyi from cxp.proto. A human runs it after changing the proto. The CXP codec imports cxp_pb2 at runtime, so the output is a runtime input; see the cxp module.

It takes no arguments. main requires uv and a GIL-enabled CPython 3.14 (the PROTOC_PYTHON path, resolved by _default_protoc_python), because grpcio-tools has no free-threaded wheel; it exits 1 with a message when either is missing. _protoc_command builds the ephemeral-environment protoc invocation, which runs in a temporary directory. The results are then published into the proto directory with dump_text_atomic from orchestrata_ultimate.core.atomic_json, so a concurrent importer never reads a half-written module. Finally it runs ruff check --fix and black over the proto directory, because protoc and mypy-protobuf emit legacy Optional and Union annotations and unsorted imports that the repository's lint gate rejects. It prints a one-line confirmation on success.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/gen_cxp.py

#!/usr/bin/env python3
"""Regenerate the CXP protobuf message stubs from cxp.proto.

Emits ``cxp_pb2.py`` + ``cxp_pb2.pyi`` only. The gRPC service stubs
(``cxp_pb2_grpc.py``) are no longer generated: the ``grpc.aio`` CXP transport
was retired on 2026-08-07 (SPEC-023 -- its EventEngine threads re-enable the
GIL) in favour of the cxpt direct transport, which speaks the same protobuf
messages over its own framing. protobuf is a codec and was never coupled to
the RPC runtime, so the messages stay and the service stubs go. See
``plans/grpc-retirement-design.md``.

**protoc runs in an ephemeral, GIL-enabled interpreter.** ``grpcio-tools`` is
deliberately NOT a project dependency:

  - it ships no ``cp314t`` wheel (checked 2026-08-07: 1.83.0 publishes
    ``cp314`` only), so depending on it meant vendoring a hand-built
    free-threaded wheel under ``vendor/wheels/`` -- which is gitignored, so a
    fresh clone (i.e. CI) could not resolve it and ``uv sync --dev`` failed;
  - it never needs to be free-threaded in the first place. protoc is a code
    generator invoked as a SUBPROCESS; it emits text and exits. Nothing about
    the GIL in that child process can affect this project's runtime.

So it is fetched on demand into a throwaway environment built on the system
GIL-enabled 3.14 interpreter. uv caches the wheels, so repeat runs are fast
and offline.

The grpcio-tools version is PINNED because protoc's output carries a gencode
version that the installed ``protobuf`` runtime validates: a runtime older
than the gencode is a hard ``VersionError`` at import. 1.80.0 emits gencode
6.31.1, which the pinned ``protobuf>=4.25`` runtime accepts. (The system
``/usr/bin/protoc`` 35.1 emits gencode 7.35.1 and is therefore unusable here
-- verified 2026-08-07.) Bumping this pin means checking the generated
``cxp_pb2.py`` still imports.

Run from the repo root:

    uv run python scripts/gen_cxp.py
"""

from __future__ import annotations

import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

PROTO_DIR = Path("src/orchestrata_ultimate/cxp/proto")
PROTO = PROTO_DIR / "cxp.proto"

# See the module docstring: pinned so the emitted gencode version stays
# compatible with the `protobuf` runtime this project pins.
GRPCIO_TOOLS_PIN = "grpcio-tools==1.80.0"


# A GIL-enabled interpreter for the throwaway protoc env. protoc is a
# subprocess code generator, so the GIL is irrelevant to it -- and the
# free-threaded build has no grpcio-tools wheel.
def _default_protoc_python() -> str:
    """Resolve a GIL-enabled CPython 3.14 to host protoc/grpcio-tools.

    Prefers the conventional system interpreter (present on the project's
    reference Debian hosts, where this is a no-op). Falls back to a
    uv-managed, exactly-pinned build on hosts that lack it (verified
    2026-09-23: a dev host here has no ``/usr/bin/python3.14`` and never
    will via its native package manager -- it is not Debian). A bare
    ``--python 3.14`` request is not a substitute: uv resolves that
    ambiguous version against whichever build is already active for this
    project (free-threaded, per ``requires-python``) rather than the
    GIL-enabled one grpcio-tools needs -- verified empirically, not assumed
    -- so the fallback pins the exact patch and asks uv for it by full
    version instead.
    """
    system_python = Path("/usr/bin/python3.14")
    if system_python.exists() or shutil.which("uv") is None:
        return str(system_python)
    pin = "3.14.3"
    try:
        subprocess.run(
            ["uv", "python", "install", pin],
            capture_output=True,
            text=True,
            check=False,
        )
        found = subprocess.run(
            ["uv", "python", "find", pin], capture_output=True, text=True, check=False
        )
    except OSError:
        return str(system_python)
    candidate = found.stdout.strip()
    if found.returncode == 0 and candidate and Path(candidate).exists():
        return candidate
    return str(system_python)


PROTOC_PYTHON = _default_protoc_python()


def _protoc_command(out_dir: str) -> list[str]:
    """Build the ephemeral-environment protoc invocation."""
    return [
        "uv",
        "run",
        "--isolated",
        "--no-project",
        "--python",
        PROTOC_PYTHON,
        "--with",
        GRPCIO_TOOLS_PIN,
        "python",
        "-m",
        "grpc_tools.protoc",
        "-I",
        str(PROTO_DIR),
        f"--python_out={out_dir}",
        f"--pyi_out={out_dir}",
        str(PROTO),
    ]


def main() -> int:
    if shutil.which("uv") is None:
        print(
            "uv is required to build the ephemeral protoc environment; "
            "see https://docs.astral.sh/uv/",
            file=sys.stderr,
        )
        return 1
    if not Path(PROTOC_PYTHON).exists():
        print(
            f"{PROTOC_PYTHON} not found -- a GIL-enabled CPython 3.14 is "
            "needed to host grpcio-tools (it ships no cp314t wheel). Install "
            "the system python3.14 package or edit PROTOC_PYTHON.",
            file=sys.stderr,
        )
        return 1

    with tempfile.TemporaryDirectory() as td:
        rc = subprocess.run(_protoc_command(td), capture_output=True, text=True)
        if rc.returncode != 0:
            print(rc.stderr, file=sys.stderr)
            return 1
        tmp = Path(td)
        # Atomic, and here it matters more than for a data artifact: these two
        # are runtime_input — the CXP codec IMPORTS cxp_pb2. Truncating the
        # module in place keeps its inode, so a concurrent importer or a
        # `git update-index` hashing it can read a half-written module or die
        # with SIGBUS. Publishing by replace gives them a new inode and leaves
        # any existing reader on the intact previous file.
        from orchestrata_ultimate.core.atomic_json import dump_text_atomic

        dump_text_atomic(PROTO_DIR / "cxp_pb2.py", (tmp / "cxp_pb2.py").read_text())
        dump_text_atomic(PROTO_DIR / "cxp_pb2.pyi", (tmp / "cxp_pb2.pyi").read_text())

    # protoc/mypy-protobuf emit legacy `Optional[...]`/`Union[...]` annotations
    # and unsorted imports in the .pyi stub, which trip the repo's ruff gate
    # (UP045/UP007/I001). Auto-fix them so regenerated files stay lint-clean
    # rather than reintroducing 50+ pre-existing errors on every regen.
    subprocess.run(
        ["uv", "run", "ruff", "check", "--fix", "-q", str(PROTO_DIR)],
        check=False,
    )
    subprocess.run(["uv", "run", "black", "-q", str(PROTO_DIR)], check=False)
    print("regenerated cxp_pb2.py, cxp_pb2.pyi")
    return 0


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