orchestrata_ultimate.agents.runtime_main

Reference › orchestrata_ultimate › agents › runtime_main

runtime_main is the entry point of the out-of-process Agent Runtime, the agent-runtime CXP gRPC server. main runs _serve, which resolves the listen port with _resolve_port (it reads CXP_GRPC_PORT or the port in CXP_GRPC_ENDPOINT, has no hardcoded default and raises a configuration error when missing), builds the packet handler with build_kernel_packet_handler from kernel_handler, creates the CXP server, and connects the handler's reply sink to the server's reply queue. The handler import is lazy so that importing the module stays light.

SIGTERM and SIGINT start _graceful_stop, which drains in-flight background rounds before the transport stops. The server lifecycle is run through run_serve_lifecycle, and _flush_audit_trail waits for pending audit writes before the event loop closes. main accepts an optional argument list, environment mapping and replacement serve coroutine for testing.

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/agents/runtime_main.py

"""Agent Runtime gRPC server entry point.

Provides a gRPC-based agent runtime service using the CXP protocol.
Used by Docker container orchestrata-agent-runtime.
"""

import asyncio
import logging
import signal
import sys
from collections.abc import Callable, Coroutine, Mapping
from typing import Any

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port
from orchestrata_ultimate.cxp.servicer import (
    CXPServerConfig,
    run_serve_lifecycle,
)
from orchestrata_ultimate.cxp.transport import create_cxp_server

logger = logging.getLogger(__name__)


def _resolve_port(env: Mapping[str, str] | None = None) -> int:
    """Resolve the CXP gRPC listen port from the environment.

    Delegates to :func:`cxp_grpc_port`, which reads ``CXP_GRPC_PORT`` (or the
    port embedded in ``CXP_GRPC_ENDPOINT``). There is **no hardcoded default** —
    the port lives in ``.env``; a missing value raises ``CXPConfigError``.

    Args:
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    return cxp_grpc_port(env=env)


async def _flush_audit_trail() -> None:
    """Wait for in-flight fire-and-forget audit writes before the loop closes.

    This process is the busiest producer of them (volk registration at boot,
    the dispatcher's per-packet audit hooks), and every one is scheduled from
    a SYNCHRONOUS call site — so nothing else awaits them. Returning from
    :func:`_serve` closes the loop and ``asyncio.run``'s task cancellation
    kills whatever is still writing, losing exactly the shutdown-window
    records a compliance trail most needs.

    ``audit_bootstrap`` is imported lazily here for the same reason
    ``kernel_handler`` is: it costs ~270ms of this module's ~500ms import,
    and only the shutdown path needs it.
    """
    from orchestrata_ultimate.kernel.audit_bootstrap import drain_audit_tasks

    drained = await drain_audit_tasks()
    if drained:
        logger.info("Drained %d in-flight audit write(s) on shutdown", drained)


async def _graceful_stop(handler: Any, server: Any) -> None:
    """Drain in-flight background rounds before stopping the transport
    (plans/kernel-dispatch-reply-hardening.md §4.2 D-H7).

    Awaits ``handler.shutdown()`` (:meth:`_KernelPacketHandler.shutdown`)
    FIRST so a still-connected consumer receives each in-flight round's
    terminal packet -- the genuine reply, or an honest
    ``intent_failed``/``intent_no_reply`` terminal -- THEN stops the gRPC
    server. The reverse order (stopping the transport first) is what let
    every deploy-time restart cancel in-flight rounds silently before this
    fix.

    Installed as the SIGTERM/SIGINT signal handler in :func:`_serve`.

    ``server.stop()`` runs in a ``finally`` (F-1 hardening) so a raising
    drain (e.g. a bug in a volk's cancellation handling) still releases the
    listening socket instead of leaving the transport running forever.
    """
    try:
        await handler.shutdown()
    finally:
        await server.stop()


async def _serve(env: Mapping[str, str] | None = None) -> None:
    """Start the agent runtime gRPC server.

    The listen port is resolved from the environment (``CXP_GRPC_PORT`` or
    ``CXP_GRPC_ENDPOINT`` in ``.env``); there is no hardcoded default, so the
    kernel always binds the single configured port the dispatcher/harness use.

    Args:
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
    )

    # Imported lazily so `from ...runtime_main import _resolve_port` stays light
    # (this pulls in the full volk engine via PipelineRunner).
    from orchestrata_ultimate.agents.kernel_handler import build_kernel_packet_handler

    port = _resolve_port(env=env)
    config = CXPServerConfig(host="0.0.0.0", port=port)
    handler = build_kernel_packet_handler()
    server = create_cxp_server(config, packet_handler=handler)
    handler.set_reply_sink(server.servicer.enqueue_reply)

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(
            sig, lambda: asyncio.ensure_future(_graceful_stop(handler, server))
        )

    # `run_serve_lifecycle` guarantees `server.stop()` runs before this
    # coroutine returns or raises -- see its docstring for why that matters
    # (a leaked half-stopped `grpc.aio.Server` otherwise hangs the whole
    # process at interpreter exit once grpc's `shutdown_grpc_aio` trips on
    # it). See ``knowledge_graph/server.py::_serve`` for the same fix
    # applied to the KG gRPC server.
    try:
        await run_serve_lifecycle(server, label="Agent Runtime server", port=port)
    finally:
        await _flush_audit_trail()


def main(
    argv: list[str] | None = None,
    env: Mapping[str, str] | None = None,
    serve: Callable[..., Coroutine[Any, Any, None]] | None = None,
) -> None:
    """Entry point for agent runtime server.

    Args:
        argv: Command-line arguments (unused, for compatibility).
        env: Optional explicit environment mapping. When provided, values are
            read from this mapping instead of ``os.environ``.
        serve: The server coroutine function to run; defaults to
            :func:`_serve`. Injection seam so the ``KeyboardInterrupt``
            handling below is testable — forcing a real SIGINT into
            ``_serve()``'s synchronous prelude races an in-flight kernel
            boot (real Redis/graph/Prolog connects) and leaves background
            threads running past the test (measured 2026-07-10).
    """
    # RUL-019: load the two-phase env contract before anything resolves an
    # infrastructure endpoint. This process is started BOTH by systemd (which
    # injects EnvironmentFile=) and from a bare shell; without this call the
    # bare-shell path had no CXP_GRPC_PORT at all and died in _resolve_port.
    # Skipped when the caller injects an explicit env mapping — that mapping
    # is then the authority and must not be overwritten from disk.
    if env is None:
        load_env_contract()
    serve_fn = _serve if serve is None else serve
    try:
        asyncio.run(serve_fn(env=env))
    except KeyboardInterrupt:
        logger.info("Agent Runtime server interrupted")
        sys.exit(0)


if __name__ == "__main__":
    main()