orchestrata_ultimate.kernel.cli_main

Reference › orchestrata_ultimate › kernel › cli_main

The orchestrata command line and the monolithic deployment entry point. The module does two jobs. Its start command boots every runtime component in one process; its other commands are thin clients that call a running system through the gateway. The package's __main__ module calls main, and it is the largest module of the kernel package. It serves SPEC-000 (the CLI and boot stories US-002, US-004 to US-007) and SPEC-001 (the heartbeat and status wiring). The curated runtime services page covers the deployed units.

Boot order in _boot_monolithic, which installs SIGTERM and SIGINT handlers and later stops components in reverse order:

• _boot_create_components starts the OrchestrationKernel, the Postman broker (see the postman package) and the state manager, and registers the latter two with the kernel. Kernel events such as kernel.boot.components, kernel.ready and kernel.stopping go out through _emit_kernel_event, which logs and never raises.
• _boot_intent_pipeline creates a Prolog bridge, loads the governance rule files, then builds the volk registry with a status bridge and a heartbeat pump.
• _boot_plugin_system and _boot_conversation_consumer add the plugin manager and the conversation consumer.
• _boot_intent_broker builds the SPEC-003 intent broker from the postman transport, the registry and the Prolog bridge, and registers it with the kernel as a discoverable component.
• _boot_agent_runtime boots a PipelineRunner, wraps it with build_kernel_packet_handler from the agents package, and serves it with a CXP server on the port from cxp_grpc_port(), the CXP_GRPC_PORT setting. The docstring describes this as mirroring the out-of-process agent runtime (runtime_main). The environment key kept the name "GRPC" after the gRPC retirement moved CXP to the cxpt direct transport.
• _boot_start_services runs a small FastAPI health server on a daemon thread with /health/live and /health/ready (the second returns 503 until the ready event is set).

The click group cli carries the client commands. intent calls the gateway method orchestrator.intent; respond, approve and reject send a cxp.send carrying a subject of human-response, human-approval or human-rejection; packets calls cxp.status; governance check calls governance.evaluate; watch streams events through cli_watch. All use cli_client. _load_config reads the YAML named by the option or ORCHESTRATA_CONFIG and expands environment references through _resolve_env_vars. Other commands cover init, stop, status, health, team, agents, review, verify, query and session.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 23 of 29 imports and 1 of 1 importers.

API

Classes

_HeartbeatPumpComponent

Line 204

Lifecycle adapter so the SPEC-001 roster HeartbeatPump stops cleanly alongside the other monolithic components (which expect async stop() + get_metadata().name).

_CXPKernelComponent

Line 345

Lifecycle adapter so the in-process CXP kernel server stops with the other monolithic components (which expect async stop() + get_metadata().name).

_KernelIntentBrokerComponent

Line 374

Lifecycle + discovery adapter holding the SPEC-003 IntentBroker so it is constructed ONCE at boot, discoverable via the kernel component registry (get_component('intent-broker')), and closed…

_PluginManagerComponent

Line 445

Lifecycle adapter so the SPEC-015 plugin system stops/unloads its plugins cleanly alongside the other monolithic components (which expect async stop() + get_metadata().name).

_ConversationConsumerComponent

Line 551

Lifecycle adapter so the SPEC-021 conversation-capture consumer runs a final flush and stops cleanly alongside the other monolithic components (which expect async stop() +…

Functions

Source

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

# mypy: ignore-errors
"""Orchestrata CLI — monolithic deployment entry point.

Commands:
  start   Boot all components in a single process
  version Show version string
  health  Check health endpoint

Author: Mohamed
"""

# Spec: 000 - Orchestrata CLI

import asyncio
import builtins
import functools
import json
import logging
import os
import re
import signal
import threading
import time
from collections.abc import Mapping
from datetime import UTC
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import click
import yaml

from orchestrata_ultimate.core.base.traces import traces

__version__ = "0.1.0"

logger = logging.getLogger("orchestrata")


@traces(spec="000", story="US-002", reqs=["REQ-000-016"])
def _load_config(
    config_path: str | None, *, env: Mapping[str, str] | None = None
) -> dict[str, Any]:
    """Load YAML config from file or ORCHESTRATA_CONFIG env var.

    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 scripts/code_intel.py's ORCHESTRATA_ROOT
    # guard — without it the shipped .env makes a no-config invocation crash.
    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:
        raw = yaml.safe_load(f) or {}
    return _resolve_env_vars(raw)


_ENV_PATTERN = re.compile(r"\${([^}^{]+)}")


@traces(spec="000", story="US-002", reqs=["REQ-000-016"])
def _resolve_env_vars(obj: Any) -> Any:
    """Recursively resolve ${VAR:-default} patterns in config values."""
    if isinstance(obj, str):

        def _replace(match: re.Match[str]) -> str:
            expr = match.group(1)
            if ":-" in expr:
                var, default = expr.split(":-", 1)
                return os.environ.get(var, default)
            return os.environ.get(expr, match.group(0))

        return _ENV_PATTERN.sub(_replace, obj)
    elif isinstance(obj, dict):
        return {k: _resolve_env_vars(v) for k, v in obj.items()}
    elif isinstance(obj, __import__("builtins").list):
        return [_resolve_env_vars(item) for item in obj]
    return obj


def _setup_logging(level: str, fmt: str = "text") -> None:
    """Configure logging with level and format."""
    import json as _json

    numeric_level = getattr(logging, level.upper(), logging.INFO)

    if fmt == "json":

        class JSONFormatter(logging.Formatter):
            def format(self, record) -> str:
                return _json.dumps(
                    {
                        "timestamp": self.formatTime(record),
                        "level": record.levelname,
                        "logger": record.name,
                        "message": record.getMessage(),
                    }
                )

        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logging.root.handlers.clear()
        logging.root.addHandler(handler)
    else:
        logging.basicConfig(
            level=numeric_level,
            format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
        )
    logging.root.setLevel(numeric_level)


@traces(spec="000", story="US-002", reqs=["REQ-000-015"])
def _start_health_server(
    host: str, port: int, ready_event: threading.Event
) -> threading.Thread:
    """Start FastAPI health server on a daemon thread."""
    import uvicorn
    from fastapi import FastAPI
    from fastapi.responses import JSONResponse

    app = FastAPI(title="Orchestrata Health", docs_url=None, redoc_url=None)

    @app.get("/health/live")
    @traces(spec="000", story="US-002", reqs=["REQ-000-015"])
    async def liveness() -> dict:
        return {"status": "alive"}

    @app.get("/health/ready", response_model=None)
    @traces(spec="000", story="US-002", reqs=["REQ-000-015"])
    async def readiness() -> dict | JSONResponse:
        if ready_event.is_set():
            return {"status": "ready"}
        return JSONResponse({"status": "starting"}, status_code=503)

    # http="h11": "auto" would import httptools, which re-enables the GIL
    # process-wide (SPEC-023) — fatal in the kernel, which runs janus.
    thread = threading.Thread(
        target=uvicorn.run,
        args=(app,),
        kwargs={"host": host, "port": port, "log_level": "warning", "http": "h11"},
        daemon=True,
    )
    thread.start()
    return thread


@traces(spec="014", story="US-012", reqs=["REQ-014-192"])
async def _emit_kernel_event(
    kernel: Any,
    event_type: str,
    message: str,
    *,
    severity: str = "info",
    **payload: Any,
) -> None:
    """Emit a ``SystemEvent`` through the kernel's EventBus, tolerating any
    failure (Redis down, EventBus misconfigured, ...) — boot/shutdown must
    NEVER die on an event-emission failure, so errors are only logged."""
    try:
        from orchestrata_ultimate.core.protocols import SystemEvent

        event = SystemEvent(
            type=event_type,
            source="kernel",
            severity=severity,
            payload={"message": message, **payload},
        )
        await kernel.emit_event(event)
    except Exception as exc:
        logger.warning("Failed to emit kernel event %s: %s", event_type, exc)


@traces(spec="000", story="US-002", reqs=["REQ-000-009", "REQ-000-010"])
async def _boot_create_components(
    config: dict[str, Any], components: builtins.list
) -> tuple:
    from orchestrata_ultimate.core.state import StateManager
    from orchestrata_ultimate.kernel import OrchestrationKernel
    from orchestrata_ultimate.postman import Postman

    kernel = OrchestrationKernel(config=config.get("core", {}).get("kernel", {}))
    postman = Postman(config=config.get("core", {}).get("postman", {}))
    state_mgr = StateManager(config=config.get("state", {}))

    for comp in (kernel, postman, state_mgr):
        await comp.initialize(config)
        await comp.start()
        components.append(comp)
        logger.info("Started: %s", comp.get_metadata().name)

    for comp in (postman, state_mgr):
        await kernel.register_component(comp)

    return kernel, postman, state_mgr


class _HeartbeatPumpComponent:
    """Lifecycle adapter so the SPEC-001 roster ``HeartbeatPump`` stops
    cleanly alongside the other monolithic components (which expect
    ``async stop()`` + ``get_metadata().name``)."""

    def __init__(self, pump: Any) -> None:
        self._pump = pump
        self._metadata = SimpleNamespace(name="volk-heartbeat-pump")

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    async def stop(self) -> None:
        self._pump.stop()

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def get_metadata(self) -> Any:
        return self._metadata


@traces(spec="000", story="US-002", reqs=["REQ-000-011"])
@traces(
    spec="001",
    story="US-001",
    reqs=["REQ-001-014", "REQ-001-018", "REQ-001-009"],
)
async def _boot_intent_pipeline(
    config: dict[str, Any], components: builtins.list, kernel: Any
) -> tuple:
    """Boot the real ``VolkRegistry`` (the intent-classification path),
    wired for SPEC-001 volk-comms-heartbeat awareness (packet P8, decision 7
    of ``plans/volk-comms-heartbeat-design.md`` §4.1 -- this is one of the
    two ``VolkRegistry`` construction sites the design names as owning that
    wiring): a ``VolkStatusEventBridge`` re-emits the registry's kernel-side
    status transitions as ``volk.status_changed`` kernel events through
    *kernel*'s existing ``_emit_kernel_event`` seam (closes G3d, REQ-001-014/
    009/022), and a ``HeartbeatPump`` beats every roster alias on a fixed
    interval so an idle-but-alive in-process volk never lazily decays to
    INACTIVE (closes G1, REQ-001-018/017). Both start once the registry is
    live and stop on the monolithic shutdown path via the returned
    ``_HeartbeatPumpComponent``/registry lifecycle adapters appended to
    *components*.
    """
    prolog_bridge = None
    registry = None
    try:
        from orchestrata_ultimate.agents.volk_registry import VolkRegistry
        from orchestrata_ultimate.core.paths import get_prolog_dir
        from orchestrata_ultimate.governance._strict import governance_strict
        from orchestrata_ultimate.governance.prolog_client import (
            GOVERNANCE_RULE_FILES,
            GovernancePrologClient,
            count_governance_predicates,
            load_governance_rules,
        )
        from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
        from orchestrata_ultimate.kernel.heartbeat_pump import HeartbeatPump
        from orchestrata_ultimate.kernel.volk_status_bridge import (
            VolkStatusEventBridge,
        )

        prolog_cfg = config.get("prolog", {})
        # Default to the CANONICAL rules dir (governance.pl, routing.pl,
        # intent_policy.pl, ...) rather than a relative "prolog" that rarely
        # exists at the CWD — otherwise the broker's validator queries an
        # undefined permission_granted/2 and fail-closed-denies every intent.
        rules_dir = Path(prolog_cfg.get("rules_dir") or get_prolog_dir())

        prolog_bridge = PrologBridge(rules_dir=rules_dir)
        if rules_dir.exists():
            # Load the canonical governance corpus through the ONE blessed
            # loader first (W1-P3) -- this is the same bridge instance the
            # intent broker's IntentValidator (permission_granted/2 governance
            # gate) and the agent-runtime kernel's SPEC-003 Option-A gate
            # (_boot_agent_runtime -> build_kernel_packet_handler) receive
            # below, so it must carry the full governance rule set, not just
            # whatever this glob happens to load first. The glob loop below
            # then picks up the non-governance-domain files this pipeline
            # also needs (routing.pl, intent_grammar.pl, ...); PrologBridge
            # tracks consulted paths per-instance, so re-globbing a file
            # already loaded here is a no-op, not a double-consult.
            loaded = load_governance_rules(prolog_bridge, rules_dir)
            predicate_count = count_governance_predicates(prolog_bridge, rules_dir)
            strict = governance_strict()
            # W1-P6 boot health line: a rule base that loaded 0 files must
            # be unmistakable here, not a silent no-op discovered later by
            # forensic audit (the exact failure class this instrumentation
            # exists to surface -- reports/proposal-wire-prolog-governance.md).
            log_level = (
                logging.WARNING if loaded == 0 or predicate_count == 0 else logging.INFO
            )
            logger.log(
                log_level,
                "governance_boot surface=cli_main._boot_intent_pipeline "
                "files_loaded=%d/%d predicates=%d strict=%s",
                loaded,
                len(GOVERNANCE_RULE_FILES),
                predicate_count,
                strict,
            )
            for pl_file in sorted(rules_dir.glob("*.pl")):
                prolog_bridge.load_rules(pl_file)
                logger.info("Loaded Prolog rules: %s", pl_file.name)

        bridge = VolkStatusEventBridge(
            emit=functools.partial(_emit_kernel_event, kernel)
        )
        # Bind THIS coroutine's running loop so on_status_change still
        # delivers when fired from a thread with no loop of its own -- the
        # production case, since HeartbeatPump beats (and so drives
        # INACTIVE->ACTIVE revivals) on its own daemon threading.Thread, not
        # this async kernel boot path. Without this, every off-thread
        # transition was silently dropped (packet P8 fix).
        bridge.bind_loop(asyncio.get_running_loop())
        # The registry is what makes a volk KNOWN to `governance:may_execute/2`.
        # That predicate denies any agent it has no volk_exists/1 fact for
        # (the permissive catch-all was deleted 2026-08-10 once these facts had
        # a producer), so without this wire every stage dispatch would be
        # refused. `prolog_bridge` is the same instance that just consulted the
        # governance corpus above, so the facts land in the rule base the gate
        # actually queries.
        registry = VolkRegistry(
            on_status_change=bridge.on_status_change,
            governance=GovernancePrologClient(bridge=prolog_bridge),
        )
        await registry.initialize(config)
        await registry.start()
        components.append(registry)
        pump = HeartbeatPump(
            probe=registry,
            roster=lambda: [reg.definition.alias for reg in registry.list_volks()],
        )
        pump.start()
        components.append(_HeartbeatPumpComponent(pump))
        logger.info("Volk registry initialized")
    except ImportError as e:
        logger.warning("Intent pipeline not available: %s", e)
    except Exception as e:
        logger.warning("Intent pipeline init failed: %s", e)

    return prolog_bridge, registry


class _CXPKernelComponent:
    """Lifecycle adapter so the in-process CXP kernel server stops with the
    other monolithic components (which expect ``async stop()`` +
    ``get_metadata().name``).

    Composes the kernel packet handler (plans/kernel-dispatch-reply-
    hardening.md §4.2 D-H7) so ``stop()`` drains in-flight background rounds
    BEFORE the transport goes down -- the monolithic boot's restart-as-
    deploy path otherwise cancelled them silently."""

    def __init__(self, server: Any, handler: Any) -> None:
        self._server = server
        self._handler = handler

    @traces(spec="000", story="US-002", reqs=["REQ-000-026"])
    async def stop(self) -> None:
        # server.stop() runs in a `finally` (F-1 hardening) so a raising
        # drain still releases the listening socket instead of leaving the
        # transport running forever.
        try:
            await self._handler.shutdown()
        finally:
            await self._server.stop(grace=1)

    @traces(spec="000", story="US-002", reqs=["REQ-000-026"])
    def get_metadata(self) -> Any:
        return SimpleNamespace(name="cxp-kernel")


class _KernelIntentBrokerComponent:
    """Lifecycle + discovery adapter holding the SPEC-003 ``IntentBroker`` so it
    is constructed ONCE at boot, discoverable via the kernel component registry
    (``get_component('intent-broker')``), and closed cleanly on shutdown.

    This is the production construction site the broker previously lacked:
    ``build_intent_broker`` was called only from tests, so the whole SPEC-003
    intent pipeline was unreachable in a running system. The broker's NL-intent
    entry (``broker.process_text``) runs alongside — it does not replace — the
    existing PipelineRunner path."""

    def __init__(self, broker: Any) -> None:
        self.broker = broker
        self._metadata = SimpleNamespace(id="intent-broker", name="intent-broker")

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
    async def stop(self) -> None:
        self.broker.close()

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
    def get_metadata(self) -> Any:
        return self._metadata


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-001"])
async def _boot_intent_broker(
    config: dict[str, Any],
    components: builtins.list,
    kernel: Any,
    postman: Any,
    prolog_bridge: Any,
    registry: Any,
) -> Any:
    """Construct the SPEC-003 ``IntentBroker`` at boot from the real Postman
    transport + Volk registry + Prolog bridge, register it as a discoverable
    kernel component, and expose its NL-intent ingress (``process_text``)
    alongside the existing PipelineRunner path.

    Returns the broker, or ``None`` if its transport/registry dependencies did
    not boot (the intent pipeline degraded)."""
    if postman is None or registry is None:
        logger.warning("Intent broker not booted: postman/registry unavailable")
        return None
    try:
        from orchestrata_ultimate.orchestrator.intent_broker_factory import (
            build_intent_broker,
        )

        redis_db = int(config.get("core", {}).get("kernel", {}).get("redis_db", 0))
        broker = build_intent_broker(
            volk_registry=registry,
            transport=postman,
            prolog_bridge=prolog_bridge,
            redis_db=redis_db,
        )
        adapter = _KernelIntentBrokerComponent(broker)
        components.append(adapter)
        await kernel.register_component(adapter)
        logger.info(
            "Intent broker booted (NL ingress: process_text) — "
            "discoverable as 'intent-broker'"
        )
        return broker
    except ImportError as e:
        logger.warning("Intent broker not available: %s", e)
        return None
    except Exception as e:
        logger.warning("Intent broker init failed: %s", e)
        return None


class _PluginManagerComponent:
    """Lifecycle adapter so the SPEC-015 plugin system stops/unloads its
    plugins cleanly alongside the other monolithic components (which expect
    ``async stop()`` + ``get_metadata().name``)."""

    def __init__(self, manager: Any) -> None:
        self._manager = manager
        self._metadata = SimpleNamespace(name="plugin-system")

    @traces(spec="015", story="US-010", reqs=["REQ-015-130"])
    async def stop(self) -> None:
        from orchestrata_ultimate.plugin.interfaces import PluginStatus

        # NOTE: ``list_plugins()`` already returns a fresh list each call — do
        # NOT wrap it in ``list(...)`` here: this module's ``@agents.command()
        # def list()`` (the "agents list" CLI command) rebinds the module-global
        # name ``list`` to a click ``Command`` object, shadowing the builtin.
        for info in self._manager.list_plugins():
            if info.status == PluginStatus.ACTIVE:
                try:
                    await self._manager.unload_plugin(info.name)
                except Exception as exc:
                    logger.error("Error unloading plugin '%s': %s", info.name, exc)

    @traces(spec="015", story="US-010", reqs=["REQ-015-130"])
    def get_metadata(self) -> Any:
        return self._metadata


@traces(spec="015", story="US-015-002", reqs=["REQ-015-017", "REQ-015-021"])
async def _boot_plugin_system(
    config: dict[str, Any],
    # NOTE: ``builtins.list`` (not bare ``list``) — this module's
    # ``@agents.command() def list()`` rebinds the module-global name
    # ``list`` to a click ``Command`` object, which pyright then rejects as
    # an annotation ("Expected class but received Command").
    components: builtins.list,
    env: Mapping[str, str] | None = None,
) -> Any:
    """Construct + boot the SPEC-015 plugin system when a plugins directory is
    configured (previously the plugin SDK was never constructed at kernel
    boot, so plugins never ran outside tests).

    Reads the plugin directory from ``ORCHESTRATA_PLUGINS_DIR`` — *env* is the
    injection seam, defaulting to ``os.environ`` at call time. Unset/empty
    means the plugin subsystem is entirely skipped: zero behavior change (a
    settled default-off posture). When set, discovers and loads every plugin
    found there through the SDK's manager/loader/discovery surface, wires
    their tool registrations into the SPEC-062 concrete ``ToolRegistry``
    (``default_registry()``), and returns the plugin ``VolkRegistry`` their
    Volks landed in so the caller can project it onto the live dispatch
    (``_boot_agent_runtime`` -> ``PipelineRunner(plugin_volk_registry=...)``,
    the seam ``tests/kernel/test_plugin_volk_projection.py`` proves).

    Out of scope (not built here): plugin Prolog rule loading, a plugin audit
    trail, a plugin CLI/ops surface, and the hot-reload manifest fix.
    """
    environ = os.environ if env is None else env
    plugins_dir = environ.get("ORCHESTRATA_PLUGINS_DIR", "")
    if not plugins_dir:
        return None
    try:
        from orchestrata_ultimate.plugin.config import PluginConfig
        from orchestrata_ultimate.plugin.interfaces import PluginStatus
        from orchestrata_ultimate.plugin.manager import PluginManager
        from orchestrata_ultimate.plugin.registries import (
            ToolRegistryAdapter,
            VolkRegistry,
        )
        from orchestrata_ultimate.tools.registry import default_registry

        plugin_cfg_dict = dict(config.get("plugins", {}))
        plugin_cfg_dict["plugin_dirs"] = [plugins_dir]
        plugin_config = PluginConfig.from_dict(plugin_cfg_dict)

        volk_registry = VolkRegistry()
        manager = PluginManager(
            config=plugin_config,
            volk_registry=volk_registry,
            tool_registry=ToolRegistryAdapter(default_registry()),
        )
        # An empty dict leaves the PluginConfig built above untouched (per
        # PluginManager.initialize) while still triggering auto-discovery.
        await manager.initialize({})
        for info in manager.list_plugins():
            if info.status == PluginStatus.DISCOVERED:
                try:
                    await manager.load_plugin(info.name)
                except Exception as exc:
                    logger.warning("Plugin '%s' failed to load: %s", info.name, exc)
        await manager.start()
        components.append(_PluginManagerComponent(manager))
        logger.info(
            "Plugin system booted from %s (%d plugin(s))",
            plugins_dir,
            len(manager.list_plugins()),
        )
        return volk_registry
    except ImportError as e:
        logger.warning("Plugin system not available: %s", e)
        return None
    except Exception as e:
        logger.warning("Plugin system init failed: %s", e)
        return None


class _ConversationConsumerComponent:
    """Lifecycle adapter so the SPEC-021 conversation-capture consumer runs a
    final flush and stops cleanly alongside the other monolithic components
    (which expect ``async stop()`` + ``get_metadata().name``)."""

    def __init__(self, consumer: Any) -> None:
        self._consumer = consumer
        self._metadata = SimpleNamespace(name="conversation-consumer")

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-070"])
    async def stop(self) -> None:
        try:
            self._consumer.flush()
        except Exception as exc:
            logger.error("Error flushing conversation consumer on stop: %s", exc)

    @traces(spec="021", story="US-021-007", reqs=["REQ-021-069"])
    def get_metadata(self) -> Any:
        return self._metadata


_CONVERSATION_CONSUMER_TRUTHY = {"1", "true", "yes", "on"}


@traces(spec="021", story="US-021-007", reqs=["REQ-021-069", "REQ-021-070"])
def _boot_conversation_consumer(
    config: dict[str, Any],
    components: builtins.list,
    env: Mapping[str, str] | None = None,
) -> Any:
    """Construct + boot the SPEC-021 conversation-capture queue consumer when
    ``ORCHESTRATA_CONVERSATION_CONSUMER`` is truthy (previously this
    subsystem was never constructed at kernel boot, so queued conversation
    turns never reached the graph outside tests).

    Reads the gate from ``ORCHESTRATA_CONVERSATION_CONSUMER`` — *env* is the
    injection seam, defaulting to ``os.environ`` at call time. Unset/falsy
    means the subsystem is entirely skipped: zero behavior change (the same
    settled default-off posture as ``ORCHESTRATA_PLUGINS_DIR``). When
    enabled, builds a real ``RedisQueueSource`` (``create_sync_redis``) and a
    real ``SyncGraphRepository`` (``create_graph_repository``, env-driven —
    REQ-021-069) into one ``ConversationQueueConsumer``, runs one boot flush
    (redelivers anything left ``processing`` from a prior crash), registers
    the ``conv_status``/``conv_sync`` tools into the default tool registry,
    and returns the consumer so the caller can append its lifecycle
    component.
    """
    environ = os.environ if env is None else env
    enabled = (
        environ.get("ORCHESTRATA_CONVERSATION_CONSUMER", "").strip().lower()
        in _CONVERSATION_CONSUMER_TRUTHY
    )
    if not enabled:
        return None
    try:
        from orchestrata_ultimate.knowledge_graph.conversation_consumer import (
            ConversationQueueConsumer,
            RedisQueueSource,
            create_conversation_graph_repository,
            create_conversation_redis,
        )
        from orchestrata_ultimate.tools.conversation_tools import (
            register_conversation_tools,
        )
        from orchestrata_ultimate.tools.registry import default_registry

        conv_cfg = config.get("conversation", {})
        queue_key = conv_cfg.get("queue_key", "orchestrata:conversation:queue")

        # The conversation queue's OWN Redis (CONVERSATION_REDIS_*), resolved
        # from *environ* (the injection seam, defaulting to os.environ at call
        # time — REQ-021-069). Never the ambient REDIS_* keys.
        redis_client = create_conversation_redis(environ)
        source = RedisQueueSource(redis_client, queue_key=queue_key)
        # The CONVERSATION graph's own endpoint. This was
        # create_graph_repository(), which resolves the profile-driven
        # KG_ROUTING_URI/KG_URI — so a kernel booted under a dev/test profile
        # flushed conversation turns into the TEST graph while every reader
        # stayed pinned to prod. See create_conversation_graph_repository.
        repo = create_conversation_graph_repository(environ)
        consumer = ConversationQueueConsumer(source=source, repo=repo)
        boot_report = consumer.flush()
        register_conversation_tools(default_registry(), consumer)
        components.append(_ConversationConsumerComponent(consumer))
        logger.info(
            "Conversation consumer booted (queue=%s): %s",
            queue_key,
            boot_report.to_summary(),
        )
        return consumer
    except ImportError as e:
        logger.warning("Conversation consumer not available: %s", e)
        return None
    except Exception as e:
        logger.warning("Conversation consumer init failed: %s", e)
        return None


@traces(spec="000", story="US-002", reqs=["REQ-000-009"])
@traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-015"])
async def _boot_agent_runtime(
    config: dict[str, Any],
    components: builtins.list,
    prolog_bridge: Any,
    registry: Any,
    intent_broker: Any = None,
    plugin_volk_registry: Any = None,
) -> None:
    """Boot the canonical CXP volk kernel (a PipelineRunner with all 9 volks) and
    serve it in-process over gRPC, mirroring the out-of-process agent-runtime
    engine (``runtime_main``). This is the engine that actually runs volks; the
    earlier in-process ``VolkRuntime`` path drove an empty registry and could
    spawn nothing (gap W6).

    The ``PipelineRunner`` this engine serves is now ALWAYS constructed and
    booted here (packet P8, closes G5) rather than only when
    *plugin_volk_registry* is supplied -- previously the no-plugin path
    passed ``kernel=None`` to :func:`build_kernel_packet_handler`, which
    built its own unwired runner internally, so the live kernel-side roster
    never carried a health probe. *registry* (the ``VolkRegistry`` built at
    ``_boot_intent_pipeline``'s construction site) is threaded in as
    ``health_probe=``, which ``PipelineRunner._register_volks`` in turn
    binds into every rostered ``BaseAgent`` volk via ``set_volk_registry``
    (REQ-001-018/015). When *plugin_volk_registry* is supplied (SPEC-015
    boot wiring) its registered Volks are additionally projected onto the
    SAME live dispatch the 10 built-in volks use."""
    try:
        from orchestrata_ultimate.agents.kernel_handler import (
            build_kernel_packet_handler,
        )
        from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port
        from orchestrata_ultimate.cxp.servicer import CXPServerConfig
        from orchestrata_ultimate.cxp.transport import create_cxp_server
        from orchestrata_ultimate.kernel.pipeline import PipelineRunner

        kernel_runner = PipelineRunner(
            plugin_volk_registry=plugin_volk_registry, health_probe=registry
        )
        kernel_runner.boot()

        port = cxp_grpc_port()
        handler = build_kernel_packet_handler(
            kernel=kernel_runner, intent_broker=intent_broker
        )
        server = create_cxp_server(
            config=CXPServerConfig(host="0.0.0.0", port=port),
            packet_handler=handler,
        )
        handler.set_reply_sink(server.servicer.enqueue_reply)
        await server.start()
        components.append(_CXPKernelComponent(server, handler))
        logger.info("CXP volk kernel listening on :%d (all 9 volks)", port)
    except ImportError as e:
        logger.warning("Agent runtime not available: %s", e)
    except Exception as e:
        logger.warning("Agent runtime init failed: %s", e)


@traces(spec="000", story="US-002", reqs=["REQ-000-015"])
def _boot_start_services(config: dict[str, Any], ready_event: threading.Event) -> None:
    api_cfg = config.get("infrastructure", {}).get("api", {})
    health_host = api_cfg.get("host", "0.0.0.0")
    health_port = api_cfg.get("port", 8000)
    _start_health_server(health_host, health_port, ready_event)
    ready_event.set()
    health_url = f"http://{health_host}:{health_port}/health/live"
    logger.info(
        "Orchestrata ready — health at %s",
        health_url,
    )


@traces(
    spec="000",
    story="US-002",
    reqs=["REQ-000-009", "REQ-000-010", "REQ-000-011", "REQ-000-015", "REQ-000-026"],
)
async def _boot_monolithic(config: dict[str, Any]) -> None:
    """Boot all components in a single process."""
    # Shutdown coordination
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop_event.set)

    ready_event = threading.Event()
    components: builtins.list = []

    kernel, postman, _state_mgr = await _boot_create_components(config, components)
    await _emit_kernel_event(
        kernel,
        "kernel.boot.components",
        "Boot components started",
        started=[c.get_metadata().name for c in components],
    )
    prolog_bridge, registry = await _boot_intent_pipeline(config, components, kernel)
    if registry is not None:
        await _emit_kernel_event(
            kernel,
            "kernel.boot.volks",
            "Volk registry started",
            volk_count=registry.count,
        )
    plugin_volk_registry = await _boot_plugin_system(config, components)
    _boot_conversation_consumer(config, components)
    broker = await _boot_intent_broker(
        config, components, kernel, postman, prolog_bridge, registry
    )
    await _boot_agent_runtime(
        config,
        components,
        prolog_bridge,
        registry,
        intent_broker=broker,
        plugin_volk_registry=plugin_volk_registry,
    )
    _boot_start_services(config, ready_event)
    await _emit_kernel_event(kernel, "kernel.ready", "Orchestrata ready")

    # --- Wait for shutdown ---
    await stop_event.wait()
    await _emit_kernel_event(
        kernel, "kernel.stopping", "Shutdown signal received", severity="warning"
    )
    logger.info("Shutdown signal received, stopping components...")

    for comp in reversed(components):
        try:
            await comp.stop()
            logger.info("Stopped: %s", comp.get_metadata().name)
        except Exception as e:
            logger.error("Error stopping %s: %s", comp.get_metadata().name, e)

    logger.info("Orchestrata shutdown complete")
    await _emit_kernel_event(kernel, "kernel.stopped", "Orchestrata shutdown complete")


# ============================================================
# CLI Group
# ============================================================


@click.group()
@traces(
    spec="000",
    story="US-001",
    reqs=[
        "REQ-000-094",
        "REQ-000-099",
        "REQ-000-100",
        "REQ-000-101",
        "REQ-000-102",
        "REQ-000-103",
    ],
)
def cli() -> None:
    """Orchestrata — Composable AI Orchestration System."""


@cli.command()
@click.option(
    "--config",
    "-c",
    "config_path",
    type=click.Path(),
    help="YAML config file path",
)
@click.option(
    "--log-level",
    default="INFO",
    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"], case_sensitive=False),
    help="Logging level",
)
@click.option(
    "--log-format",
    default="text",
    type=click.Choice(["text", "json"]),
    help="Log output format",
)
@traces(
    spec="000",
    story="US-002",
    reqs=[
        "REQ-000-009",
        "REQ-000-010",
        "REQ-000-011",
        "REQ-000-012",
        "REQ-000-013",
        "REQ-000-015",
    ],
)
def start(config_path: str | None, log_level: str, log_format: str) -> None:
    """Start Orchestrata in monolithic mode."""
    _setup_logging(log_level, log_format)
    config = _load_config(config_path)
    logger.info("Starting Orchestrata in monolithic mode...")
    asyncio.run(_boot_monolithic(config))


@cli.command()
@traces(spec="000", story="US-010", reqs=["REQ-000-095"])
def version() -> None:
    """Show Orchestrata version."""
    click.echo(f"orchestrata {__version__}")


@cli.command()
@click.option("--host", default="localhost", help="Health endpoint host")
@click.option("--port", default=8000, type=int, help="Health endpoint port")
def health(host: str, port: int) -> None:
    """Check Orchestrata health endpoint."""
    import urllib.request

    try:
        url = f"http://{host}:{port}/health/live"
        with urllib.request.urlopen(url, timeout=5) as resp:
            data = json.loads(resp.read())
            click.echo(f"Status: {data.get('status', 'unknown')}")
    except (OSError, ValueError) as e:
        click.echo(f"Health check failed: {e}", err=True)
        raise SystemExit(1)


# ── OUTPUT HELPERS ──────────────────────────────────────────────────


@traces(spec="000", story="US-010", reqs=["REQ-000-096"])
def _emit(payload: dict[str, Any], json_mode: bool, text_fn=None) -> None:
    """Emit a result as JSON (--json, REQ-000-096) or human text."""
    if json_mode:
        click.echo(json.dumps(payload, indent=2, default=str))
    elif text_fn is not None:
        text_fn(payload)
    else:
        for key, value in payload.items():
            click.echo(f"{key}: {value}")


# ── INIT COMMAND ──────────────────────────────────────────────────


VALID_VOLK_ALIASES = ("U:Ox", "U:PS", "U:Vz", "U:Ux", "U:PM", "U:Ar", "U:Qa", "U:Cx")

_PROJECT_NAME_RE = re.compile(r"^[a-z][a-z0-9-]*$")


@cli.command()
@click.argument("project")
@click.option("--template", default="microservice", help="Project template type")
@click.option(
    "--directory",
    "-C",
    "base_dir",
    type=click.Path(file_okay=False),
    default=".",
    help="Parent directory for the new project",
)
@traces(
    spec="000",
    story="US-001",
    reqs=[
        "REQ-000-001",
        "REQ-000-002",
        "REQ-000-003",
        "REQ-000-005",
        "REQ-000-006",
        "REQ-000-007",
        "REQ-000-008",
    ],
)
def init(project: str, template: str, base_dir: str) -> None:
    """Initialize a new Orchestrata project (REQ-000-001..008)."""
    from orchestrata_ultimate.kernel.cli_client import load_dotenv

    load_dotenv()

    if not _PROJECT_NAME_RE.match(project):
        raise click.ClickException(
            f"Invalid project name {project!r}: must match "
            "[a-z][a-z0-9-]* (lowercase, digits, dashes)"
        )

    root = Path(base_dir) / project
    if root.exists():
        raise click.ClickException(f"Directory already exists: {root}")

    # REQ-000-001: project directory structure
    (root / "prolog").mkdir(parents=True)
    (root / "specs").mkdir()

    # REQ-000-002: context.json (session source of truth)
    (root / "context.json").write_text(
        json.dumps(
            {
                "project": project,
                "template": template,
                "created": time_now_iso(),
                "sessions": [],
                "current_session": None,
            },
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )

    # REQ-000-003/007: orchestrata.yaml from the template
    config = {
        "project": project,
        "template": template,
        "team": {"active_volks": list(VALID_VOLK_ALIASES)},
        "runtime": {"health_port": 8000, "log_level": "INFO"},
    }
    (root / "orchestrata.yaml").write_text(
        yaml.safe_dump(config, sort_keys=False), encoding="utf-8"
    )

    # REQ-000-004: Prolog governance rules into the project
    gov_src = Path(__file__).resolve().parent.parent / "prolog" / "governance.pl"
    if gov_src.exists():
        (root / "prolog" / "governance.pl").write_text(
            gov_src.read_text(encoding="utf-8"), encoding="utf-8"
        )
    else:
        logger.warning("governance.pl not found at %s; skipping", gov_src)

    # REQ-000-005: register the project in the knowledge graph (best-effort)
    graph_registered = False
    # The marker must be the LAST line before the flagged line — the gate reads
    # only that line and the one directly above it.
    # gate-ok: graph_registered is False before the try and True only on success, so the handler informs by leaving it False; the caller reads it
    try:
        from orchestrata_ultimate.knowledge_graph.sync import GraphSyncer

        syncer = GraphSyncer()
        try:
            syncer.connect()
            syncer._connected_repo.execute_write(
                "MERGE (p:Project {slug: $slug}) "
                "SET p.name = $name, p.template = $template",
                {"slug": project, "name": project, "template": template},
            )
            graph_registered = True
        finally:
            syncer.close()
    except Exception as e:
        logger.warning("Project not registered in graph (%s); init continues", e)

    # REQ-000-006: first session opened automatically
    session_opened = False
    try:
        from orchestrata_ultimate.session.session import SessionController

        ctrl = SessionController(str(root), ai_model="orchestrata-cli")
        ctrl.begin(purpose=f"Project {project} initialized", objectives=[])
        session_opened = True
    except Exception as e:
        logger.warning("Could not open first session (%s); init continues", e)

    click.echo(f"Project initialized: {root}")
    click.echo(f"  context.json, orchestrata.yaml, prolog/ ({template})")
    click.echo(
        f"  graph registration: {'ok' if graph_registered else 'skipped (unreachable)'}"
    )
    click.echo(f"  first session: {'opened' if session_opened else 'not opened'}")


@traces(spec="000", story="US-001", reqs=["REQ-000-002"])
def time_now_iso() -> str:
    """UTC timestamp for context.json."""
    from datetime import datetime

    return datetime.now(UTC).isoformat()


# ── STOP COMMAND ───────────────────────────────────────────────────


@cli.command()
@traces(
    spec="000",
    story="US-002",
    reqs=["REQ-000-017", "REQ-000-018", "REQ-000-019", "REQ-000-020"],
)
def stop() -> None:
    """Stop the Orchestrata runtime (REQ-000-017..020).

    Archives the current session, then gracefully stops the gateway and
    kernel processes via their PID files (SIGTERM with drain, matching
    scripts/start_test_services.py semantics).
    """
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc, load_dotenv

    load_dotenv()

    # REQ-000-017: archive current session (best-effort — session end must
    # never block the shutdown path)
    try:
        from orchestrata_ultimate.session.session import SessionController

        ctrl = SessionController(".", ai_model="orchestrata-cli")
        current = ctrl.active_session
        if current is None:
            # `stop` runs in a separate process from the one that began the
            # session, so the in-memory active session is always None here —
            # restore the latest on-disk session and archive it only when it
            # was never closed (length stays None until end()).
            resumed = ctrl.resume()
            if resumed is not None and resumed.length is None:
                current = resumed
        if current is not None:
            ctrl.end(
                achievements=["Session archived by `orchestrata stop`"],
                upcoming=[],
            )
            click.echo("Session archived")
        else:
            click.echo("No active session to archive")
    except Exception as e:
        logger.warning("Session archive failed (%s); shutdown continues", e)

    # REQ-000-018..020: graceful stop of gateway + kernel via PID files.
    # Postman routing drains with the kernel (it owns the queues); Redis and
    # the graph are external services — we disconnect by stopping our side.
    # The PID-file paths are env-overridable (ordinary config, /tmp defaults
    # unchanged): the suite's test_stop injects a tmp path so a batch
    # concurrent test never SIGTERMs whatever live PID another worker's
    # start_test_services PID-file tests just wrote into the shared /tmp
    # file (the 2026-08-05 test_cli_main/test_stop batch flake — it reported
    # "stopped (PID ...)" for a dummy process owned by a different worker).
    stopped: list[str] = []
    for name, pid_file in (
        (
            "gateway",
            os.environ.get(
                "ORCHESTRATA_GATEWAY_PID_FILE", "/tmp/orchestrata-gateway.pid"
            ),
        ),
        (
            "kernel",
            os.environ.get(
                "ORCHESTRATA_KERNEL_PID_FILE", "/tmp/orchestrata-kernel.pid"
            ),
        ),
    ):
        pid = _read_pid_file(pid_file)
        if pid is None:
            click.echo(f"{name}: not running (no PID file)")
            continue
        try:
            os.kill(pid, signal.SIGTERM)
            deadline = time.monotonic() + 10
            alive = True
            while time.monotonic() < deadline:
                try:
                    os.kill(pid, 0)
                    time.sleep(0.2)
                except ProcessLookupError:
                    alive = False
                    break
            if alive:
                logger.warning("%s (PID %d) did not exit within 10s", name, pid)
            else:
                stopped.append(name)
                click.echo(f"{name}: stopped (PID {pid})")
        except ProcessLookupError:
            click.echo(f"{name}: stale PID file (PID {pid} already gone)")
            Path(pid_file).unlink(missing_ok=True)

    # Best-effort final state via gateway if it is somehow still up.
    if "gateway" not in stopped:
        try:
            gateway_rpc("gateway.ping", timeout=2.0)
            click.echo("gateway: still responding after SIGTERM — check manually")
        except click.ClickException:
            pass  # down or never up; nothing more to do


@traces(spec="000", story="US-002", reqs=["REQ-000-020"])
def _read_pid_file(path: str) -> int | None:
    """Read and validate a PID file (os.kill(pid, 0) liveness check)."""
    p = Path(path)
    if not p.exists():
        return None
    try:
        pid = int(p.read_text(encoding="utf-8").strip())
        os.kill(pid, 0)
        return pid
    except ValueError, OSError:
        return None


# ── STATUS COMMAND ─────────────────────────────────────────────────


@cli.command()
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(
    spec="000",
    story="US-002",
    reqs=[
        "REQ-000-021",
        "REQ-000-022",
        "REQ-000-023",
        "REQ-000-024",
        "REQ-000-025",
    ],
)
def status(json_mode: bool) -> None:
    """Show Orchestrata runtime status (REQ-000-021..025)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc, gateway_uri

    payload: dict[str, Any] = {"gateway": gateway_uri()}
    try:
        ping = gateway_rpc("gateway.ping", timeout=5.0)
        payload["runtime"] = "running"
        payload["ping"] = ping.get("status")
    except click.ClickException:
        payload["runtime"] = "stopped"
        _emit(payload, json_mode, _format_status)
        return

    for key, method in (
        ("volks", "volk.list"),
        ("session", "session.active"),
    ):
        try:
            payload[key] = gateway_rpc(method, timeout=5.0)
        except click.ClickException as e:
            payload[key] = {"error": str(e)}
    # Health lives on the gateway's HTTP surface (serve_http), not the old
    # FastAPI app's :8000 — resolve host/port the same way the CLI resolves
    # its gateway RPC address (cli_client) plus GATEWAY_HTTP_PORT.
    health_host = os.environ.get("GATEWAY_HOST", "localhost")
    health_port = os.environ.get("GATEWAY_HTTP_PORT", "8766")
    payload["health"] = f"http://{health_host}:{health_port}/health/live"
    _emit(payload, json_mode, _format_status)


def _format_status(p: dict[str, Any]) -> None:
    """Human-readable rendering of composite runtime status."""
    click.echo(f"Runtime state: {p.get('runtime', 'unknown')}")
    click.echo(f"Gateway: {p.get('gateway', '?')}")
    volks = (
        p.get("volks", {}).get("volks", []) if isinstance(p.get("volks"), dict) else []
    )
    click.echo(f"Active volks: {len(volks)}")
    for v in volks:
        click.echo(
            f"  {v.get('alias', '?'):8s} {v.get('status', '?')} ({v.get('health', '?')})"
        )
    session = p.get("session", {})
    if isinstance(session, dict) and session:
        click.echo(f"Session: {session.get('id', session.get('session', 'active'))}")
    else:
        click.echo("Session: none")
    if p.get("health"):
        click.echo(f"Health endpoint: {p['health']}")


# ── CONFIG COMMAND GROUP ──────────────────────────────────────────


@cli.group()
def config() -> None:
    """Configure Orchestrata settings."""


@traces(spec="000", story="US-003", reqs=["REQ-000-032"])
def _project_config_path() -> Path:
    """Locate orchestrata.yaml for the current project (cwd or ORCHESTRATA_PROJECT)."""
    base = Path(os.environ.get("ORCHESTRATA_PROJECT", "."))
    path = base / "orchestrata.yaml"
    if not path.exists():
        raise click.ClickException(
            f"No orchestrata.yaml found in {base.resolve()} — run `orchestrata init` first"
        )
    return path


@traces(spec="000", story="US-003", reqs=["REQ-000-030", "REQ-000-031"])
def _validate_volks(volks: tuple[str, ...]) -> list[str]:
    """Validate volk aliases (REQ-000-030/031); rejects invalid ones loudly."""
    invalid = [v for v in volks if v not in VALID_VOLK_ALIASES]
    if invalid:
        raise click.ClickException(
            f"Invalid volk alias(es): {', '.join(invalid)}. "
            f"Valid aliases: {', '.join(VALID_VOLK_ALIASES)}"
        )
    return list(volks)


@traces(spec="000", story="US-003", reqs=["REQ-000-029"])
def _write_team_file(active: list[str]) -> None:
    """Write prolog/team.pl with active_volk/1 facts (REQ-000-029)."""
    path = _project_config_path().parent / "prolog" / "team.pl"
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "% Active volk team — generated by `orchestrata config team`",
        "% Loaded by the runtime on start; edit via the CLI, not by hand.",
        *[f"active_volk('{v}')." for v in sorted(active)],
        "",
    ]
    path.write_text("\n".join(lines), encoding="utf-8")


@config.command(name="team")
@click.option("--add", "add_volks", multiple=True, help="Volk alias to activate")
@click.option(
    "--remove", "remove_volks", multiple=True, help="Volk alias to deactivate"
)
@traces(
    spec="000",
    story="US-003",
    reqs=[
        "REQ-000-027",
        "REQ-000-028",
        "REQ-000-029",
        "REQ-000-030",
        "REQ-000-031",
        "REQ-000-032",
    ],
)
def team(add_volks: tuple[str, ...], remove_volks: tuple[str, ...]) -> None:
    """Configure the active volk team (REQ-000-027..032).

    Example: orchestrata config team --add U:Cx --add U:Ar --remove U:Ux
    """
    if not add_volks and not remove_volks:
        raise click.ClickException("Provide --add and/or --remove with volk aliases")
    path = _project_config_path()
    cfg = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    active = list(cfg.get("team", {}).get("active_volks", []))

    added, removed = [], []
    for v in _validate_volks(add_volks):
        if v not in active:
            active.append(v)
            added.append(v)
    for v in _validate_volks(remove_volks):
        if v in active:
            active.remove(v)
            removed.append(v)

    cfg.setdefault("team", {})["active_volks"] = active
    path.write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8")
    _write_team_file(active)
    click.echo(f"Active volks: {', '.join(active) if active else '(none)'}")
    click.echo(f"Added: {', '.join(added) if added else '(none new)'}")
    click.echo(f"Removed: {', '.join(removed) if removed else '(none present)'}")


# ── AGENTS COMMAND GROUP ───────────────────────────────────────────


@cli.group()
def agents() -> None:
    """Manage and monitor agents."""


@agents.command(name="list")
@traces(spec="000", story="US-005", reqs=["REQ-000-042", "REQ-000-043"])
def agents_list() -> None:
    """List all registered volks (REQ-000-042/043)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("volk.list")
    volks = result.get("volks", [])
    if not volks:
        click.echo("No volks currently registered")
        return
    for v in volks:
        click.echo(
            f"{v.get('alias', '?'):8s} {v.get('role', '?'):14s} "
            f"status={v.get('status', '?')} health={v.get('health', '?')} "
            f"task={v.get('current_task') or '-'}"
        )


@agents.command(name="health")
@traces(spec="000", story="US-005", reqs=["REQ-000-044", "REQ-000-045"])
def agents_health() -> None:
    """Show health status of all volks (REQ-000-044/045)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("volk.list")
    volks = result.get("volks", [])
    if not volks:
        click.echo("No volks currently registered")
        return
    for v in volks:
        click.echo(f"{v.get('alias', '?'):8s} {v.get('health', 'unknown')}")


@agents.command()
@click.argument("thread_id")
@traces(
    spec="000",
    story="US-005",
    reqs=["REQ-000-046", "REQ-000-047", "REQ-000-048", "REQ-000-049"],
)
def packets(thread_id: str) -> None:
    """Show CXP packet flow for a thread (REQ-000-046..049)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("cxp.status", {"thread_id": thread_id})
    events = result.get("events", [])
    click.echo(f"Thread: {thread_id} — status: {result.get('status', 'unknown')}")
    for e in events:
        click.echo(
            f"  {e.get('from_agent', '?')} -> {e.get('to_agent', '?')} "
            f"[{e.get('subject', '-')}] "
            f"delivery={e.get('delivery_status', '?')} "
            f"work={e.get('work_status', '?')}"
        )
    if not events:
        click.echo("  (no packet traffic observed for this thread)")


# ── INTENT / RESPOND / APPROVE / REJECT / REVIEW ──────────────────────


def _format_intent_result(r: dict[str, Any]) -> None:
    """Human-readable rendering of an orchestrator.intent result."""
    click.echo(f"Intent: {r.get('intent_id', '?')}")
    if r.get("needs_disambiguation"):
        click.echo("Blocked by governance — clarification needed:")
        for reason in r.get("reasons", []):
            click.echo(f"  - {reason}")
        return
    task = r.get("task")
    volk = task.get("volk", "?") if isinstance(task, dict) else "?"
    click.echo(f"Thread: {r.get('thread_id') or '(not dispatched)'}")
    click.echo(f"Routed to: {volk}")
    click.echo(f"Summary: {r.get('summary', '')}")


@cli.command()
@click.argument("text")
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(
    spec="000",
    story="US-004",
    reqs=[
        "REQ-000-033",
        "REQ-000-034",
        "REQ-000-037",
        "REQ-000-038",
        "REQ-000-041",
    ],
)
def intent(text: str, json_mode: bool) -> None:
    """Submit a natural-language intent to the Intent Broker (REQ-000-033..041)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("orchestrator.intent", {"text": text})
    _emit(result, json_mode, _format_intent_result)


@cli.command()
@click.option("--thread", "thread_id", required=True, help="Thread ID to respond on")
@click.argument("text")
@traces(spec="000", story="US-004", reqs=["REQ-000-039", "REQ-000-040"])
def respond(thread_id: str, text: str) -> None:
    """Send a response to a volk on a thread (REQ-000-039/040)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc(
        "cxp.send",
        {
            "target": thread_id,
            "payload": {
                "thread_id": thread_id,
                "subject": "human-response",
                "text": text,
            },
        },
    )
    click.echo(
        f"Response sent on thread {thread_id} (status: {result.get('status', 'sent')})"
    )


@cli.command()
@click.option("--thread", "thread_id", required=True, help="Thread ID to approve")
@traces(
    spec="000",
    story="US-006",
    reqs=["REQ-000-053", "REQ-000-054", "REQ-000-057", "REQ-000-058"],
)
def approve(thread_id: str) -> None:
    """Approve work on a thread — work_status=resolved (REQ-000-053/054)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc(
        "cxp.send",
        {
            "target": thread_id,
            "payload": {
                "thread_id": thread_id,
                "subject": "human-approval",
                "work_status": "resolved",
            },
        },
    )
    click.echo(f"Approved thread {thread_id} (status: {result.get('status', 'sent')})")


@cli.command()
@click.option("--thread", "thread_id", required=True, help="Thread ID to reject.")
@click.argument("reason")
@traces(spec="000", story="US-006", reqs=["REQ-000-055", "REQ-000-056", "REQ-000-057"])
def reject(thread_id: str, reason: str) -> None:
    """Reject work on a thread and provide feedback (REQ-000-055/056)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc(
        "cxp.send",
        {
            "target": thread_id,
            "payload": {
                "thread_id": thread_id,
                "subject": "human-rejection",
                "work_status": "needs_work",
                "feedback": reason,
            },
        },
    )
    click.echo(
        f"Rejected thread {thread_id} with feedback (status: {result.get('status', 'sent')})"
    )


@cli.command()
@click.option("--thread", "thread_id", required=True, help="Thread ID to watch")
@click.option(
    "--count",
    default=100,
    show_default=True,
    help="Replay backlog depth",
)
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(spec="000", story="US-006", reqs=["REQ-000-096", "REQ-000-097"])
def watch(thread_id: str, count: int, json_mode: bool) -> None:
    """Stream a thread's events -- backlog then live -- one line each
    (REQ-000-096/097). THE PHASE 0 PROOF (SPEC-001 US-006): consumes the
    shipped ``PipelineEventType`` taxonomy through the gateway's WebSocket
    JSON-RPC surface (C3 -- never Redis directly), exiting 0 on a terminal
    event (``pipeline.complete``/``pipeline.cancelled``) or cleanly on
    Ctrl-C."""
    from orchestrata_ultimate.kernel.cli_watch import (
        is_terminal_event,
        render_event_line,
        stream_thread_events,
    )

    async def _run() -> None:
        async for event in stream_thread_events(thread_id, count=count):
            if json_mode:
                click.echo(json.dumps(event, default=str))
            else:
                click.echo(render_event_line(event))
            if is_terminal_event(event):
                return

    try:
        asyncio.run(_run())
    except KeyboardInterrupt:
        return


@cli.command()
@click.argument("volk")
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(spec="000", story="US-006", reqs=["REQ-000-051", "REQ-000-052"])
def review(volk: str, json_mode: bool) -> None:
    """Show a volk's output/artifacts on active threads (REQ-000-051/052)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("knowledge.volk_work_context", {"volk": volk})
    _emit(result, json_mode)


# ── GOVERNANCE COMMAND GROUP ───────────────────────────────────────


@cli.group()
def governance() -> None:
    """Query and inspect governance policies."""


@governance.command()
@click.argument("action")
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(
    spec="000",
    story="US-007",
    reqs=["REQ-000-059", "REQ-000-060", "REQ-000-061", "REQ-000-066"],
)
def check(action: str, json_mode: bool) -> None:
    """Check if an action is governance-allowed (REQ-000-059..061)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("governance.evaluate", {"intent": action})
    _emit(
        result,
        json_mode,
        lambda r: click.echo(f"{r.get('result', 'unknown')}: {r.get('reason', '')}"),
    )
    if not json_mode and result.get("result") == "denied":
        raise SystemExit(1)


@governance.command()
@traces(spec="000", story="US-007", reqs=["REQ-000-062", "REQ-000-063"])
def log() -> None:
    """Show governance decision history (REQ-000-062/063)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("governance.decisions")
    decisions = result.get("decisions", [])
    if not decisions:
        click.echo("No governance decisions found")
        return
    for d in decisions:
        click.echo(
            f"[{d.get('timestamp', '?')}] {d.get('action', '?')} -> "
            f"{'allowed' if d.get('allowed') else 'denied'} "
            f"({d.get('reason', '')}) rules={d.get('articles_checked', [])}"
        )


@governance.command()
@traces(spec="000", story="US-007", reqs=["REQ-000-064", "REQ-000-065"])
def policies() -> None:
    """List active governance policies (REQ-000-064/065)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("governance.rules")
    rules = result.get("rules", result.get("policies", []))
    if not rules:
        click.echo("No governance policies loaded")
        return
    for r in rules:
        if isinstance(r, dict):
            click.echo(
                f"- {r.get('name', r.get('id', '?'))}: {r.get('description', '')}"
            )
        else:
            click.echo(f"- {r}")


# ── VERIFY COMMAND ──────────────────────────────────────────────────


@cli.command()
@click.option("--spec", "spec_id", help="Specific spec to verify")
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(
    spec="000",
    story="US-008",
    reqs=["REQ-000-067", "REQ-000-068", "REQ-000-069", "REQ-000-070"],
)
def verify(spec_id: str | None, json_mode: bool) -> None:
    """Verify requirements coverage against specs (REQ-000-067..069)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    params = {"spec": spec_id} if spec_id else {}
    result = gateway_rpc("knowledge.rtm_progress", params)
    _emit(result, json_mode, _format_verify_result)


def _format_verify_result(r: dict[str, Any]) -> None:
    """Human-readable rendering of RTM progress.

    ``knowledge.rtm_progress`` returns either an itemized ``specs``/
    ``requirements`` list or an aggregate ``{verified, total, percentage}``
    dict. The lookup must NOT default to ``[]`` — that made the aggregate
    case silently match ``isinstance(rows, list)`` and print nothing
    (defect found while writing coverage for this command).
    """
    rows = r.get("specs", r.get("requirements"))
    if isinstance(rows, list):
        for row in rows:
            if isinstance(row, dict):
                click.echo(
                    f"{row.get('spec', row.get('id', '?')):24s} "
                    f"verified={row.get('verified', '?')} "
                    f"partial={row.get('partial', '?')} "
                    f"missing={row.get('missing', '?')}"
                )
            else:
                click.echo(str(row))
    else:
        for key, value in r.items():
            click.echo(f"{key}: {value}")


# ── KNOWLEDGE COMMAND GROUP ─────────────────────────────────────────


@cli.group()
def knowledge() -> None:
    """Query the knowledge graph."""


@knowledge.command()
@click.argument("query")
@click.option("--json", "json_mode", is_flag=True, help="Machine-readable output")
@traces(
    spec="000",
    story="US-009",
    reqs=["REQ-000-073", "REQ-000-074", "REQ-000-075", "REQ-000-076"],
)
def query(query: str, json_mode: bool) -> None:
    """Query the knowledge graph (REQ-000-073..076)."""
    from orchestrata_ultimate.kernel.cli_client import gateway_rpc

    result = gateway_rpc("knowledge.query", {"query": query})
    _emit(result, json_mode)


# ── SESSION COMMAND GROUP ───────────────────────────────────────────


@cli.group()
@traces(spec="000", story="US-010", reqs=["REQ-000-086"])
def session() -> None:
    """Manage work sessions."""
    click.echo(
        "Session management - use orchestrata-session CLI for full functionality"
    )


@traces(
    spec="000",
    story="US-001",
    reqs=["REQ-000-112"],
)
def main() -> None:
    """Entry point for console_scripts."""
    cli()


if __name__ == "__main__":
    main()