orchestrata_ultimate.kernel.cli

Reference › orchestrata_ultimate › kernel › cli

The command-line entry point for running the kernel as a standalone process. A click group cli carries start, status and version; status is a placeholder, and main is the console-script entry.

start takes a config path and a log level, calls _load_config, and runs _run_kernel under asyncio. _load_config reads the YAML file named by the option or by ORCHESTRATA_CONFIG; it returns an empty configuration when no path is set or when the value is an unsubstituted CHANGEME_ template sentinel, and raises a click error when the named file does not exist. _run_kernel installs SIGTERM and SIGINT handlers that set an asyncio.Event, builds an OrchestrationKernel from the core.kernel section of the configuration, initialises and starts it, logs the process id, waits for the stop event and then stops the kernel. This path starts only the kernel object; the full deployment with the postman and the agent pipeline boots from cli_main. _load_config carries a traces marker for SPEC-001 requirement REQ-001-US-009-001.

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.py

# mypy: ignore-errors
"""Orchestrata Kernel CLI — standalone kernel process.

Commands:
  start   Boot the kernel component
  status  Show kernel status

Author: Mohamed
"""

# Spec: 001 - Orchestration Kernel

import asyncio
import logging
import os
import signal
from collections.abc import Mapping
from pathlib import Path
from typing import Any

import click
import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel import OrchestrationKernel

logger = logging.getLogger("orchestrata_ultimate.kernel")

__version__ = "0.1.0"


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-001"])
def _load_config(
    config_path: str | None, *, env: Mapping[str, str] | None = None
) -> dict[str, Any]:
    """Load YAML config.

    Args:
        config_path: Explicit config file path.
        env: Optional environment mapping (defaults to os.environ).
    """
    active_env = os.environ if env is None else env
    path = config_path or active_env.get("ORCHESTRATA_CONFIG")
    # An unsubstituted .env template sentinel (CHANGEME_*) means "not
    # configured" — same convention as kernel/cli_main._load_config, the
    # reference implementation of the guard.
    if not path or path.startswith("CHANGEME_"):
        return {}
    p = Path(path)
    if not p.exists():
        raise click.ClickException(f"Config file not found: {path}")
    with open(p) as f:
        return yaml.safe_load(f) or {}


async def _run_kernel(config: dict[str, Any]) -> None:
    """Run the kernel as a standalone process."""
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop_event.set)

    kernel_config = config.get("core", {}).get("kernel", {})
    kernel = OrchestrationKernel(config=kernel_config)
    await kernel.initialize(config)
    await kernel.start()

    logger.info("Kernel started — PID %d", os.getpid())
    await stop_event.wait()

    logger.info("Kernel shutting down...")
    await kernel.stop()
    logger.info("Kernel stopped")


@click.group()
def cli() -> None:
    """Orchestrata Kernel — Central coordination engine."""


@cli.command()
@click.option(
    "--config", "-c", "config_path", type=click.Path(), help="YAML config file"
)
@click.option(
    "--log-level",
    default="INFO",
    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"], case_sensitive=False),
)
def start(config_path: str | None, log_level: str) -> None:
    """Start the kernel as a standalone process."""
    logging.basicConfig(
        level=getattr(logging, log_level.upper(), logging.INFO),
        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
    )
    config = _load_config(config_path)
    asyncio.run(_run_kernel(config))


@cli.command()
def status() -> None:
    """Show kernel status (placeholder)."""
    click.echo("kernel: not connected (use 'orchestrata health' for running instances)")


@cli.command()
def version() -> None:
    """Show kernel version."""
    click.echo(f"orchestrata-kernel {__version__}")


def main() -> None:
    """Entry point for console_scripts."""
    cli()


if __name__ == "__main__":
    main()