orchestrata_ultimate.kernel.cli_client

Reference › orchestrata_ultimate › kernel › cli_client

The gateway client helpers for the orchestrata command line. Commands that talk to a running system do not open Redis or the kernel directly; they call the gateway's JSON-RPC interface over a WebSocket, and this module is that client. cli_main and cli_watch import it. It serves SPEC-000 story US-004 (REQ-000-033). The server side is the gateway server, and the token handling uses gateway.auth.

load_dotenv runs the shared two-phase environment loader (core.env_bootstrap.load_env_contract) once per process, so a bare shell resolves the same endpoints as the services. gateway_uri resolves the address in this order: CXP_WS_URI; then GATEWAY_HOST and GATEWAY_PORT, with localhost and 8765 as defaults; an explicit environment mapping can be passed for tests and then skips the loader. _rpc_call connects, sends auth.authenticate with a short-lived token (one hour, from _mint_token), and then sends the one requested JSON-RPC method and returns its result as a dictionary. gateway_rpc is the synchronous wrapper with a 15-second timeout. It turns every failure into a click error without a traceback: an authentication or method error, or an unreachable gateway, in which case the message tells the user to start the runtime (REQ-000-097).

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/cli_client.py

"""Gateway client helpers for the ``orchestrata`` CLI.

The CLI is a thin client over the running runtime (SPEC-000): interaction
commands talk to the gateway's WebSocket JSON-RPC surface instead of
booting subsystems in-process. This module centralizes:

- ``.env`` loading (same walk-up pattern as the gateway CLI),
- gateway URI resolution from env (never hardcoded hosts),
- short-lived JWT minting for ``auth.authenticate``,
- a synchronous ``gateway_rpc`` helper usable from Click commands.

Author: Mohamed
"""

# Spec: 000 - Orchestrata CLI

from __future__ import annotations

import asyncio
import json
import os
import time
from collections.abc import Mapping
from typing import Any

import click

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.env_bootstrap import load_env_contract

_dotenv_loaded = False


def load_dotenv() -> None:
    """Load the RUL-019 two-phase env contract, once per process.

    Delegates to the ONE canonical loader
    (``core.env_bootstrap.load_env_contract``); this used to be a private
    ``.env``-only walk-up that skipped the ``env/<profile>.env`` overlay, so
    the CLI resolved the prod-BASE endpoints in ``.env`` from a bare shell.
    The idempotence guard stays: the overlay writes with override, and
    re-running it would clobber a value a caller exported deliberately after
    the first load.
    """
    global _dotenv_loaded
    if _dotenv_loaded:
        return
    load_env_contract()
    _dotenv_loaded = True


def gateway_uri(env: Mapping[str, str] | None = None) -> str:
    """Resolve the gateway WebSocket URI from the environment.

    Order: CXP_WS_URI, then GATEWAY_HOST/GATEWAY_PORT, then the
    conventional localhost:8765 (a default, not a hardcoded service IP).

    Args:
        env: Optional explicit environment mapping (mirrors
            ``gateway.cli.run_serve``'s ``env=`` seam). When provided,
            resolution reads this mapping instead of ``os.environ`` and
            skips the ``.env`` load — tests pass a hermetic mapping and
            must not mutate the real process environment.
    """
    if env is None:
        load_dotenv()
        env = os.environ
    uri = env.get("CXP_WS_URI", "")
    if uri:
        return uri
    host = env.get("GATEWAY_HOST", "localhost")
    port = env.get("GATEWAY_PORT", "8765")
    return f"ws://{host}:{port}"


def _mint_token(subject: str = "orchestrata-cli", ttl_seconds: int = 3600) -> str:
    """Mint a short-lived JWT for auth.authenticate (HS256, shared secret)."""
    import jwt

    from orchestrata_ultimate.gateway.auth import TokenAuthenticator

    secret = TokenAuthenticator._get_secret()
    now = int(time.time())
    return jwt.encode(
        {"sub": subject, "iat": now, "exp": now + ttl_seconds},
        secret,
        algorithm="HS256",
    )


@traces(spec="000", story="US-004", reqs=["REQ-000-033"])
async def _rpc_call(
    uri: str, method: str, params: dict[str, Any] | None, timeout: float
) -> dict[str, Any]:
    """Connect, authenticate, call one JSON-RPC method, return the result."""
    import websockets

    async with websockets.connect(uri, close_timeout=timeout) as ws:
        await ws.send(
            json.dumps(
                {
                    "jsonrpc": "2.0",
                    "method": "auth.authenticate",
                    "params": {"token": _mint_token()},
                    "id": 0,
                }
            )
        )
        auth_resp = json.loads(await asyncio.wait_for(ws.recv(), timeout=timeout))
        if "error" in auth_resp:
            raise click.ClickException(
                f"gateway authentication failed: "
                f"{auth_resp['error'].get('message', auth_resp['error'])}"
            )
        await ws.send(
            json.dumps(
                {
                    "jsonrpc": "2.0",
                    "method": method,
                    "params": params or {},
                    "id": 1,
                }
            )
        )
        resp = json.loads(await asyncio.wait_for(ws.recv(), timeout=timeout))
        if "error" in resp:
            err = resp["error"]
            raise click.ClickException(f"{method} failed: {err.get('message', err)}")
        result = resp.get("result", {})
        return result if isinstance(result, dict) else {"value": result}


@traces(spec="000", story="US-004", reqs=["REQ-000-033"])
def gateway_rpc(
    method: str,
    params: dict[str, Any] | None = None,
    timeout: float = 15.0,
    env: Mapping[str, str] | None = None,
) -> dict[str, Any]:
    """Call a gateway JSON-RPC method synchronously.

    Raises a clean ClickException (no traceback, REQ-000-097) when the
    gateway is unreachable, the call errors, or the reply is not a dict.

    Args:
        env: Optional explicit environment mapping forwarded to
            :func:`gateway_uri` (see its docstring) — used by tests to
            target a hermetic gateway instance without mutating
            ``os.environ``.
    """
    uri = gateway_uri(env=env)
    try:
        return asyncio.run(_rpc_call(uri, method, params, timeout))
    except click.ClickException:
        raise
    except (TimeoutError, OSError) as e:
        raise click.ClickException(
            f"gateway unreachable at {uri} ({e}) — is the runtime up? "
            f"Start it with `orchestrata start` or scripts/start_test_services.py."
        ) from e
    except Exception as e:
        raise click.ClickException(f"{method} failed: {e}") from e