orchestrata_ultimate.gateway.server

Reference › orchestrata_ultimate › gateway › server

server is the composition root of the gateway: WebSocketGateway builds every subsystem, owns their lifecycle and accepts the WebSocket connections (SPEC-014). The gateway speaks JSON-RPC over WebSocket only, on port 8765 by default; it never serves HTML, because the VizInt UI is served separately on port 5000. cli starts it, and the package __init__ re-exports it. The package narrative is in gateway.

• Construction. __init__ creates the authenticator (auth_handler), the session connection manager, the pubsub manager, the ring_buffer, the audit writer, the rpc handler and the protocol message processor.
• Start-up. start tolerates a missing Redis or graph. It connects Redis and the graph, creates the graph repository and the test_runner, registers the memory engine and policy engine, wires the kernel, starts the event forwarders and the config service, opens the WebSocket server, starts the heartbeat monitor and serves the HTTP surface from http_surface. stop shuts these down in order.
• Kernel link. _init_kernel reads the CXP endpoint settings through cxp.endpoint_config and makes bounded, backed-off attempts through _connect_grpc_kernel_once, which creates the client with cxp.transport.create_cxp_client (the cxpt direct transport), connects it and runs a health check. On success the kernel mode is recorded as grpc and _consume_kernel_replies starts. If the kernel is unconfigured or unreachable, the gateway falls back to the in-process PipelineKernelClient and records mode in-process, or unavailable if even that fails. The mode is passed to RPCHandler.set_kernel_status, so gateway.info can tell the cases apart, and kernel_reprobe retries in the background.
• Connections. _handle_connection authenticates each socket, registers it, then passes every frame to _handle_message. That refreshes the heartbeat, sends JSON-RPC 2.0 requests to _handle_jsonrpc and other frames to the message processor. RPC errors are returned with their code, and an unexpected failure returns an internal error carrying a correlation id; error shapes come from error_taxonomy. On close the socket's subscriptions and record are removed and an audit event is written.
• Heartbeat. The library's own keepalive is disabled; _heartbeat_loop pings every connection on the configured interval and disconnects one after the configured number of consecutive missed pongs.
• Pumps. _pump_kernel_replies_once broadcasts kernel replies on cxp.traffic; _pump_pipeline_events_once tails the pipeline Redis stream and broadcasts on pipeline.<thread_id>; _pump_kernel_events_once forwards kernel lifecycle events on kernel.events; _pump_config_events_once forwards config.changed with secret values masked. Each has a looping _consume_* task.

VizIntWebUI is an alias subclass of WebSocketGateway. HandshakeNoiseFilter quiets handshake log noise.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 22 of 33 imports and 2 of 2 importers.

API

Classes

HandshakeNoiseFilter

Bases: logging.Filter · Line 67

Collapse the bare-TCP-probe handshake traceback to one debug line.

WebSocketGateway

Line 374

Main WebSocket gateway server.

VizIntWebUI

Bases: WebSocketGateway · Line 1459

WebUI alias for WebSocketGateway.

Functions

Source

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

"""WebSocket gateway server -- ties all subsystems together.

Implements pure JSON-RPC 2.0 message handling with proper error
codes and support for both JSON-RPC and legacy WSMessage auth.
"""

import asyncio
import json
import logging
import os
import time
import uuid
from typing import TYPE_CHECKING, Any, cast

import websockets

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph.config import GraphRepositoryConfig

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph.adapters.bolt_async import (
        BoltAsyncRepository,
    )
from orchestrata_ultimate.core.logging import correlation_id, new_correlation_id
from orchestrata_ultimate.core.protocols.memory import IMemoryEngine
from orchestrata_ultimate.di.container import DIContainer
from orchestrata_ultimate.gateway.audit import AuditWriter
from orchestrata_ultimate.gateway.auth_handler import (
    authenticate,
    create_authenticator,
    try_connect_graph,
    try_connect_redis,
)

try:
    # The explicit neo4j import keeps GRAPH_DRIVER_AVAILABLE honest: the factory
    # module may already be cached in sys.modules, so importing it alone
    # would not raise when the neo4j package itself is unavailable.
    # GraphRepositoryConfig (imported unconditionally above) is a plain
    # dataclass with no neo4j dependency, so it does not need this guard --
    # only the adapter factory (which imports the neo4j driver) does.
    import neo4j  # noqa: F401

    from orchestrata_ultimate.core.graph import create_async_graph_repository

    GRAPH_DRIVER_AVAILABLE = True
except ImportError:
    # Optional-import fallback: the None sentinel deliberately shadows the
    # factory's callable type (mypy's documented optional-dependency idiom).
    create_async_graph_repository = None  # type: ignore[assignment]
    GRAPH_DRIVER_AVAILABLE = False
from orchestrata_ultimate.gateway.config import GatewayConfig
from orchestrata_ultimate.gateway.error_taxonomy import ErrorCategory, error_data
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.kernel_reprobe import KernelReprober
from orchestrata_ultimate.gateway.protocol import MessageProcessor
from orchestrata_ultimate.gateway.pubsub import PubSubManager
from orchestrata_ultimate.gateway.ring_buffer import RedisRingBuffer
from orchestrata_ultimate.gateway.rpc import RPCHandler
from orchestrata_ultimate.gateway.session import ConnectionManager
from orchestrata_ultimate.gateway.test_runner import TestRunner
from orchestrata_ultimate.memory.memory_engine import MemoryEngine

logger = logging.getLogger(__name__)


class HandshakeNoiseFilter(logging.Filter):
    """Collapse the bare-TCP-probe handshake traceback to one debug line.

    Every connect-and-close liveness probe (start_test_services.py port
    checks, promote's verify arm, anything that opens the port and hangs
    up) makes websockets log a multi-frame InvalidMessage <- EOFError
    traceback at ERROR on the ``websockets.server`` logger — 46 of them in
    12h in the app container journal, burying real errors in the exact way
    that hid the gateway stop() AttributeError. A probe is not an error;
    log one DEBUG line and drop the record.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        exc = record.exc_info[1] if record.exc_info else None
        if isinstance(exc, websockets.exceptions.InvalidMessage):
            logger.debug("non-HTTP connection attempt (liveness probe): %s", exc)
            return False
        return True


_HANDSHAKE_NOISE_FILTER = HandshakeNoiseFilter()

JSONRPC_PARSE_ERROR = -32700
JSONRPC_INVALID_REQUEST = -32600
JSONRPC_METHOD_NOT_FOUND = -32601
JSONRPC_INVALID_PARAMS = -32602
JSONRPC_INTERNAL_ERROR = -32603


@traces(spec="014", story="US-011", reqs=["REQ-014-173"])
def _jsonrpc_error(
    msg_id: Any,
    code: int,
    message: str,
    data: dict[str, Any] | None = None,
) -> str:
    """Build a JSON-RPC 2.0 error response string.

    Args:
        msg_id: Request identifier (may be None).
        code: JSON-RPC error code.
        message: Human-readable error message.
        data: Optional additional error data.

    Returns:
        JSON-encoded error response.
    """
    # SPEC-098 FR-006/FR-007: every error carries a code, a message and a
    # `details` object, and the code is categorised AUTH|AUTHZ|VAL|RES|RATE|SYS.
    # The numeric JSON-RPC code stays exactly as it was — it is protocol-level
    # and clients switch on it — so the category is ADDITIVE, in `data`. It
    # answers a different question: whether to re-authenticate, fix the
    # request, back off, or give up. A caller that knows the category passes
    # it in `data["category"]`; otherwise it is derived from the code.
    supplied = dict(data or {})
    explicit = supplied.pop("category", None)
    err: dict[str, Any] = {"code": code, "message": message}
    err["data"] = error_data(
        code,
        category=ErrorCategory(explicit) if explicit else None,
        details=supplied,
    )
    return json.dumps({"jsonrpc": "2.0", "id": msg_id, "error": err})


@traces(spec="014", story="US-011", reqs=["REQ-014-173"])
def _jsonrpc_result(msg_id: Any, result: Any) -> str:
    """Build a JSON-RPC 2.0 success response string.

    Args:
        msg_id: Request identifier.
        result: Result payload.

    Returns:
        JSON-encoded success response.
    """
    return json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result})


async def _pump_kernel_replies_once(
    client: Any,
    broadcast_fn: Any,
    *,
    agent_id: str = "HUMAN",
    timeout_seconds: int = 5,
) -> int:
    """Drain one window of volk replies from the kernel's streaming channel and
    broadcast each to UI subscribers on ``cxp.traffic``. Returns the count.

    Separated from the gateway so it is testable against a real kernel without a
    full server: ``broadcast_fn`` is any ``async (topic, payload) -> Any`` (the
    gateway passes its own :meth:`WebSocketGateway.broadcast`).
    """
    replies = await client.receive_packets(
        agent_id=agent_id, timeout_seconds=timeout_seconds
    )
    for reply in replies:
        await broadcast_fn("cxp.traffic", reply)
    return len(replies)


async def _pump_pipeline_events_once(
    redis_client: Any,
    broadcast_fn: Any,
    last_id: str,
    *,
    block_ms: int = 1000,
    count: int = 100,
    stream_key: str | None = None,
) -> tuple[str, int]:
    """Forward one window of kernel pipeline events to UI subscribers.

    Tails the global ``gw:ringbuf:pipeline.events`` stream (written by
    :meth:`PipelineEventBus._forward_to_ui`) starting after ``last_id`` and
    broadcasts each event on its per-thread topic ``pipeline.<thread_id>`` so
    the CXPFeed shows real-time stage transitions. Cross-process: the kernel may
    run behind gRPC but shares this Redis stream.

    Separated from the gateway so it is testable against real Redis without a
    full server. ``broadcast_fn`` is any ``async (topic, payload) -> Any``.

    Args:
        redis_client: Async Redis client (the gateway's ring-buffer client).
        broadcast_fn: Async ``(topic, payload)`` broadcaster.
        last_id: Stream id to read after (``"$"`` for new-only on first call).
        block_ms: Blocking read timeout in milliseconds.
        count: Max events per window.
        stream_key: Stream to tail; defaults to the global
            ``PIPELINE_STREAM_KEY``. Injection seam so tests can listen on a
            test-unique stream — asserting "no new events" on the SHARED key
            races any residual background writer in the same process/DB
            (order-dependent batch flake, full-run 5, 2026-07-10).

    Returns:
        ``(new_last_id, forwarded_count)``.
    """
    from orchestrata_ultimate.kernel.pipeline_events import (
        PIPELINE_STREAM_KEY,
        pipeline_topic,
    )

    key = PIPELINE_STREAM_KEY if stream_key is None else stream_key
    response = await redis_client.xread({key: last_id}, count=count, block=block_ms)
    if not response:
        return last_id, 0
    forwarded = 0
    new_last = last_id
    for _, entries in response:
        for entry_id, fields in entries:
            new_last = entry_id
            payload = _decode_pipeline_event(fields)
            thread_id = payload.get("thread_id", "")
            if thread_id:
                await broadcast_fn(pipeline_topic(thread_id), payload)
                forwarded += 1
    return new_last, forwarded


async def _pump_kernel_events_once(
    redis_client: Any,
    last_id: str,
    broadcast_fn: Any,
    *,
    count: int = 100,
    block: int = 1000,
    stream_key: str | None = None,
) -> str:
    """Forward one window of kernel lifecycle events to UI subscribers.

    Tails the ``EVENTS_STREAM_KEY`` Redis stream (XADD'd by
    ``EventBus._publish_to_redis`` on every ``OrchestrationKernel.emit_event``
    call) starting after ``last_id`` and broadcasts each entry on
    ``kernel.events`` as a FLAT payload matching the vizint-ui
    NotificationCenter's vocabulary (``type``/``priority``/``message``/
    ``timestamp``) -- never nested like the governance forwarder.

    Separated from the gateway so it is testable against real Redis without a
    full server. ``broadcast_fn`` is any ``async (topic, payload) -> Any`` (the
    gateway passes its own :meth:`WebSocketGateway.broadcast`).

    Args:
        redis_client: Async Redis client (the gateway's ring-buffer client).
        last_id: Stream id to read after (``"$"`` for new-only on first call).
        broadcast_fn: Async ``(topic, payload)`` broadcaster.
        count: Max events per window.
        block: Blocking read timeout in milliseconds.
        stream_key: Stream to tail; defaults to the shared
            ``EVENTS_STREAM_KEY``. Injection seam so tests can exercise
            error paths on a test-unique key instead of clobbering the
            shared stream.

    Returns:
        The new ``last_id``.
    """
    from orchestrata_ultimate.kernel.event_bus import EVENTS_STREAM_KEY

    key = EVENTS_STREAM_KEY if stream_key is None else stream_key
    response = await redis_client.xread({key: last_id}, count=count, block=block)
    if not response:
        return last_id
    new_last = last_id
    for _, entries in response:
        for entry_id, fields in entries:
            new_last = entry_id
            payload = _decode_kernel_event(fields)
            await broadcast_fn("kernel.events", payload)
    return new_last


def _decode_kernel_event(fields: dict[Any, Any]) -> dict[str, Any]:
    """Normalise a kernel-events stream entry to the FLAT UI payload the
    vizint-ui NotificationCenter reads (``type``/``priority``/``message``/
    ``timestamp``), mirroring ``_decode_pipeline_event``.

    Stream fields may come back as bytes (no ``decode_responses``) or str; the
    ``payload`` field is a JSON string written by ``EventBus._publish_to_redis``.
    """

    def _s(value: Any) -> str:
        return value.decode("utf-8") if isinstance(value, bytes) else str(value)

    decoded: dict[str, Any] = {_s(k): _s(v) for k, v in fields.items()}
    raw_payload = decoded.get("payload", "{}")
    try:
        payload = json.loads(raw_payload) if raw_payload else {}
    except json.JSONDecodeError, TypeError:
        payload = {}

    payload = dict(payload)
    message = payload.pop("message", None)
    severity = decoded.get("severity", "info")

    return {
        "type": decoded.get("type", ""),
        "severity": severity,
        "priority": severity,
        "message": message,
        "timestamp": decoded.get("timestamp", ""),
        "source": decoded.get("source", ""),
        "correlation_id": decoded.get("correlation_id") or None,
        "payload": payload,
    }


def _decode_pipeline_event(fields: dict[Any, Any]) -> dict[str, Any]:
    """Normalise a Redis-stream pipeline-event entry to a UI payload dict.

    Stream fields may come back as bytes (no ``decode_responses``) or str; the
    ``data`` field is JSON-decoded back to a dict.
    """

    def _s(value: Any) -> str:
        return value.decode("utf-8") if isinstance(value, bytes) else str(value)

    decoded: dict[str, Any] = {_s(k): _s(v) for k, v in fields.items()}
    raw_data = decoded.get("data", "{}")
    try:
        decoded["data"] = json.loads(raw_data) if raw_data else {}
    except json.JSONDecodeError, TypeError:
        decoded["data"] = {}
    try:
        decoded["iteration"] = int(decoded.get("iteration", 0))
    except TypeError, ValueError:
        decoded["iteration"] = 0
    return decoded


async def _pump_config_events_once(
    store: Any,
    broadcast_fn: Any,
    last_id: str = "$",
) -> int:
    """Forward one window of runtime-config change events via *broadcast_fn*
    on topic ``config.changed``, redacting secret values first (SPEC-014
    REQ-014-083). Separated from the gateway so it is testable against real
    Redis without a full server -- mirrors ``_pump_pipeline_events_once``.

    Args:
        store: A connected ``RuntimeConfigStore``.
        broadcast_fn: Any ``async (topic, payload) -> Any`` broadcaster.
        last_id: Forwarded to ``store.watch_events`` -- overrides the
            store's own remembered cursor (tests use this to isolate a run
            to a captured tail position); real callers pass nothing.

    Returns:
        The number of ``config.changed`` events forwarded.
    """
    from orchestrata_ultimate.gateway.runtime_config import SecretRedactor

    redactor = SecretRedactor()
    changes = await store.watch_events(last_id)
    forwarded = 0
    for change in changes:
        value = "***" if redactor.is_secret_key(change.key) else change.value
        await broadcast_fn(
            "config.changed",
            {
                "key": change.key,
                "value": value,
                "actor": change.actor,
                "timestamp": change.timestamp,
            },
        )
        forwarded += 1
    return forwarded


class WebSocketGateway:
    """Main WebSocket gateway server.

    Coordinates authentication, connection management,
    pub/sub, ring buffer, audit, and RPC dispatch.
    """

    def __init__(self, config: GatewayConfig) -> None:
        """Build all subsystem instances.

        Args:
            config: Gateway configuration.
        """
        self._config = config
        self._auth = create_authenticator(config)
        self._connections = ConnectionManager(config)
        self._pubsub = PubSubManager()
        self._ring_buffer = RedisRingBuffer(config)
        self._audit = AuditWriter(config)
        self._ws_map: dict[str, Any] = {}
        self._rpc = RPCHandler(
            self._connections,
            self._pubsub,
            self._ring_buffer,
            self._audit,
            session_root=config.session_root,
        )
        self._server: Any = None
        self._processor = MessageProcessor(
            self._rpc,
            self._connections,
            self._ws_map,
            self._pubsub,
        )
        # Wire processor back to RPC for broadcast support
        self._rpc.set_processor(self._processor)
        # Test runner (created after processor so broadcast works)
        self._test_runner: Any = None
        # Live CXP kernel client + its reply-consumer task (wired in start()).
        self._kernel_client: Any = None
        self._reply_task: Any = None
        # Which path _init_kernel took: "grpc" | "in-process" | "unavailable".
        # `self._kernel_client is not None` alone is NOT sufficient to tell
        # these apart -- the in-process PipelineKernelClient fallback also
        # leaves _kernel_client set, so this is the only place that
        # distinguishes "connected to the real kernel" from "quietly running
        # the fallback" (surfaced on gateway.info per the "kernel" contract).
        self._kernel_mode: str = "unavailable"
        self._kernel_endpoint: str | None = None
        # Kernel re-probe (SPEC-014, gateway-kernel-reprobe contract): the
        # resolved (host, port) so the re-prober never re-reads the
        # environment, the in-process client retained across a switch to grpc
        # (closing it would cancel in-flight pipelines), and the re-prober.
        self._grpc_target: tuple[str, int] | None = None
        self._fallback_client: Any = None
        self._kernel_reprober: KernelReprober | None = None
        # P-B3 (design D6): this gateway's minted reply-consumer id, stamped
        # onto outbound chat packets and used to key this gateway's private
        # reply queue on the kernel's ReplyRouter (P-B1/P-B2) -- closes the
        # 50/50 reply loss (G3) when multiple gateways share one kernel.
        self._reply_consumer_id: str = f"HUMAN@{uuid.uuid4().hex[:12]}"
        # Pipeline-event forwarder task (kernel PipelineEventBus → UI topics).
        self._pipeline_task: Any = None
        # Kernel-events forwarder task (OrchestrationKernel.emit_event → UI
        # topic "kernel.events", read by vizint-ui's NotificationCenter).
        self._kernel_events_task: Any = None
        # Stream-key injection seams for the two forwarders: None means the
        # shared production streams (PIPELINE_STREAM_KEY / EVENTS_STREAM_KEY,
        # resolved lazily inside the pump helpers). Tests override these with
        # a test-unique key so error-path tests never clobber shared streams.
        self._pipeline_stream_key: str | None = None
        self._kernel_events_stream_key: str | None = None
        # SPEC-014 US-014-005: runtime-config store + its config:events
        # forwarder task (config.changed → UI topics). Wired in start(),
        # torn down in stop().
        self._config_store: Any = None
        self._config_task: Any = None
        # SPEC-014 FR-014: server-initiated heartbeat ping loop, started in
        # start(), cancelled in stop(). Owns the only pinger on served
        # connections -- see _start_websocket_server for why the websockets
        # library's own built-in keepalive is disabled.
        self._heartbeat_task: Any = None
        # Gateway-owned DI container + the singleton memory engine it hosts
        # (wired in start()). One engine is shared by the memory RPC handler and
        # the session-close cold-migration path (SPEC-010 FR-001/FR-008).
        self._di: DIContainer | None = None
        self._memory_engine: MemoryEngine | None = None
        # Operational-graph repository used by knowledge/session/governance
        # handlers (wired in start()), kept here so stop() can close it --
        # RPCHandler only forwards it to sub-handlers and does not retain a
        # reference itself.
        self._graph_repository: Any = None

    @traces(spec="010", story="US-001", reqs=["REQ-010-002"])
    def _register_memory_engine(self) -> None:
        """Build the one gateway-owned memory engine, register it as the
        ``IMemoryEngine`` singleton in the gateway DI container (SPEC-024
        adoption / SPEC-010 FR-001), and inject that same instance into the
        memory RPC handler and the session-close cold-migration path so both
        see the same hot store.

        Construction does not connect — the engine lazily connects on first
        use, so this adds nothing to startup and tolerates missing Redis/graph
        exactly like the rest of ``start()``.
        """
        self._di = DIContainer(name="gateway")
        self._memory_engine = MemoryEngine()
        # IMemoryEngine is now MemoryEngineProtocol (interfaces->protocols
        # collapse), which MemoryEngine structurally satisfies directly — no
        # cast needed now that both trees share one MemoryEntry hierarchy.
        # DI registration KEYED by the protocol type is the point — mypy's
        # type-abstract check assumes the type will be instantiated, which a
        # container key never is.
        self._di.register_instance(IMemoryEngine, self._memory_engine)  # type: ignore[type-abstract]
        self._rpc.set_memory_engine(self._memory_engine)
        logger.info("Memory engine registered (DI singleton) and wired to handlers")

    # -- lifecycle -------------------------------------------

    async def _init_redis(self) -> None:
        """Connect the ring buffer to Redis (tolerates it being unavailable)."""
        await try_connect_redis(self._ring_buffer)

    async def _init_graph_audit(self) -> None:
        """Connect the audit writer to graph (tolerates it being unavailable)."""
        await try_connect_graph(self._audit)

    def _init_graph_repository(self) -> Any:
        """Build the operational-graph repository used by knowledge/session/
        governance handlers and wire it into the RPC handler.

        Returns:
            The connected repository, or ``None`` if construction failed.
        """
        graph_repository = None
        if create_async_graph_repository is None:
            return graph_repository
        try:
            graph_repository = create_async_graph_repository(
                GraphRepositoryConfig(
                    uri=self._config.graph_uri,
                    auth=(self._config.graph_user, self._config.graph_password),
                )
            )
            # Force eager driver construction (matches the old
            # create_async_graph semantics): a malformed URI must fail here,
            # at start(), not silently defer to the first query. ``.driver``
            # is a documented escape hatch on the concrete adapter, not on
            # the AsyncGraphRepository protocol -- cast to reach it, per
            # plans/neo4j-seam-design.md.
            _ = cast("BoltAsyncRepository", graph_repository).driver
            self._rpc.set_graph_repository(graph_repository)
            self._graph_repository = graph_repository
            logger.info("Graph repository injected for knowledge queries")
        except Exception:
            logger.warning("Graph repository unavailable for knowledge queries")
            graph_repository = None
        return graph_repository

    def _init_test_runner(self, graph_repository: Any) -> None:
        """Wire the test runner with broadcast capability."""
        try:
            self._test_runner = TestRunner(
                broadcast_fn=self.broadcast,
                graph_repository=graph_repository,
                project_root=".",
            )
            self._rpc.set_test_runner(self._test_runner)
            logger.info("Test runner initialised")
        except Exception:
            logger.exception("Test runner unavailable")

    def _init_policy_engine(self) -> None:
        """Wire the governance PolicyEngine for gateway dry-run evaluation +
        the active rule set. Registered as a deferred factory so the Prolog
        (janus) engine is not spun up until governance is actually evaluated
        -- keeping gateway startup (and the gateway test suite) free of janus
        threads.
        """

        def _make_policy_engine() -> Any:
            from orchestrata_ultimate.governance.policy_engine import PolicyEngine
            from orchestrata_ultimate.governance.prolog_client import (
                GovernancePrologClient,
                load_governance_rules,
            )
            from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

            # SPEC-007 defect fix: without consulting the governance rule
            # files here, constitutional_check/3, ethical_harm_check/3, etc.
            # are undefined predicates -- every constitutional/ethical query
            # silently falls back to its permissive default and enforcement
            # is a no-op. load_governance_rules() consults governance.pl,
            # constitutional.pl, agents.pl, ethics.pl, pam_rules.pl,
            # resources.pl, and completion.pl into this bridge's engine.
            bridge = PrologBridge()
            load_governance_rules(bridge)
            return PolicyEngine(prolog_client=GovernancePrologClient(bridge))

        self._rpc.set_policy_engine_factory(_make_policy_engine)
        logger.info("Governance PolicyEngine factory registered")
        # Admin token gating volk.register/deregister (unset -> admin ops denied).
        self._rpc.set_admin_token(self._config.admin_token)

    async def _probe_kernel_live(self) -> bool:
        """Whether the out-of-process CXP kernel answers RIGHT NOW.

        Handed to the handler layer so ``gateway.info``'s kernel block reports
        current truth instead of a boot-time outcome. Bounded by the handler;
        any failure -- wedged, closed, or torn down -- reads as not live,
        which is the honest answer.
        """
        client = self._kernel_client
        if client is None or self._kernel_mode != "grpc":
            return False
        try:
            await client.health_check()
            return True
        except Exception:
            return False

    @traces(
        spec="SPEC-014",
        story="US-014-005",
        reqs=["REQ-014-002", "REQ-014-071", "REQ-014-075"],
    )
    async def _connect_grpc_kernel_once(self, host: str, port: int) -> str | None:
        """One bounded attempt to wire the real CXP kernel.

        The single place that knows how to wire the gRPC kernel: shared by the
        boot probe loop in ``_init_kernel`` and the background re-probe.
        Returns ``None`` on success or the failure reason; never raises. On
        failure ``self._kernel_client`` is left untouched (the caller owns it).
        """
        kernel_client = None
        try:
            from orchestrata_ultimate.cxp.proto_codec import CXPClientConfig
            from orchestrata_ultimate.cxp.transport import create_cxp_client

            kernel_client = create_cxp_client(
                config=CXPClientConfig(host=host, port=port)
            )
            # Bounded: the cxpt transport dials a real TCP endpoint in
            # native code, so connect() is not free. Against a wedged
            # kernel (accept queue full, 2026-08-05) an unbounded
            # connect() black-holed the event loop until
            # pytest-timeout's 120s kill.
            ready_timeout = self._config.grpc_ready_timeout_seconds
            await asyncio.wait_for(kernel_client.connect(), timeout=ready_timeout)
            # gRPC channels connect lazily: connect() alone never proves
            # the kernel is actually serving. Verify with a bounded health
            # check so an unreachable endpoint (e.g. a dead CXP_GRPC_PORT
            # in .env with no kernel running) falls through to the
            # in-process PipelineKernelClient instead of accepting
            # intents that would only fail at send_packet time.
            await asyncio.wait_for(kernel_client.health_check(), timeout=ready_timeout)
            self._rpc.set_kernel_client(kernel_client)
            self._rpc.set_reply_consumer_id(self._reply_consumer_id)
            self._kernel_client = kernel_client
            self._kernel_mode = "grpc"
            self._kernel_endpoint = f"{host}:{port}"
            # Liveness probe, not just the boot outcome: a mode
            # latched at startup would keep answering
            # "wired: true" long after the kernel died, and the
            # one check whose job is spotting kernel-less
            # operation would pass on that stale value.
            self._rpc.set_kernel_status(
                self._kernel_mode,
                self._kernel_endpoint,
                self._probe_kernel_live,
            )
            self._reply_task = asyncio.create_task(self._consume_kernel_replies())
            return None
        except Exception as exc:
            if kernel_client is not None:
                try:
                    # Bounded for the same reason as connect() above: a
                    # cxpt disconnect against a wedged endpoint stalled
                    # the loop for the full pytest-timeout budget.
                    await asyncio.wait_for(kernel_client.disconnect(), timeout=5.0)
                except Exception:
                    logger.debug("gRPC client cleanup after failed probe")
            return str(exc)

    @traces(
        spec="SPEC-014",
        story="US-014-011",
        reqs=["REQ-014-071", "REQ-014-138"],
    )
    async def _reprobe_grpc_kernel(self) -> bool:
        """``KernelConnectAttempt`` for the re-prober: try the kernel once more.

        The in-process fallback client stays referenced in
        ``self._fallback_client`` (closing it would cancel in-flight
        pipelines) and is closed in ``stop()``.
        """
        if self._kernel_mode == "grpc":
            return True
        if self._grpc_target is None:
            return False
        host, port = self._grpc_target
        return (await self._connect_grpc_kernel_once(host, port)) is None

    @traces(spec="SPEC-014", story="US-014-011", reqs=["REQ-014-166"])
    def _init_kernel_reprobe(self) -> None:
        """Start the background re-probe when running on the fallback.

        Only when the in-process fallback is installed AND a kernel endpoint is
        configured; ``unavailable`` and unconfigured never re-probe.
        """
        if self._kernel_mode != "in-process" or self._grpc_target is None:
            return
        from orchestrata_ultimate.core.tracing.tracer import Tracer

        host, port = self._grpc_target
        interval = self._config.grpc_reprobe_interval_seconds
        reprober = KernelReprober(
            self._reprobe_grpc_kernel,
            interval_seconds=interval,
            endpoint=f"{host}:{port}",
            tracer=Tracer.default(self._config.environ),
        )
        self._kernel_reprober = reprober
        if reprober.start() is not None:
            logger.info(
                "kernel re-probe scheduled every %.1fs for %s:%d", interval, host, port
            )
        else:
            logger.info("kernel re-probe disabled (GATEWAY_GRPC_REPROBE_INTERVAL_S=0)")

    async def _init_kernel(self) -> None:
        """Wire the live CXP volk kernel (out-of-process agent-runtime), or
        fall back to the in-process ``PipelineKernelClient`` when gRPC is
        unavailable.

        Records the outcome on ``self._kernel_mode``/``self._kernel_endpoint``
        and forwards it to ``RPCHandler.set_kernel_status`` so ``gateway.info``
        can distinguish "connected to the real kernel" from "quietly running
        the in-process fallback" -- before this, both left
        ``self._kernel_client is not None``, so nothing downstream could tell
        the two states apart.

        Four distinct outcomes, each logged at the right severity:
          1. Unconfigured: neither CXP_GRPC_PORT nor CXP_GRPC_ENDPOINT in .env
             -> falls through to the in-process fallback.
          2. Unreachable: env set but the kernel never became ready within
             ``grpc_connect_attempts`` bounded, backed-off retries (closes the
             boot race where After= orders the kernel unit's SPAWN, not its
             bind of CXP_GRPC_PORT) -> falls through to the in-process
             fallback.
          3. Connected: gRPC channel opened for intent dispatch -> mode "grpc".
          4. Even the in-process fallback failed to construct -> mode
             "unavailable".
        """
        self._kernel_client = None
        self._kernel_mode = "unavailable"
        self._kernel_endpoint = None
        environ = (
            self._config.environ if self._config.environ is not None else os.environ
        )
        try:
            from orchestrata_ultimate.cxp.endpoint_config import (
                cxp_grpc_host,
                cxp_grpc_port,
            )
        except ImportError:
            logger.info("CXP gRPC client not available; skipping kernel")
        else:
            try:
                host = cxp_grpc_host(environ)
                port = cxp_grpc_port(environ)
            except Exception as exc:
                logger.info(
                    "CXP kernel not configured (%s); " "intents run in task-only mode",
                    exc,
                )
            else:
                attempts = max(1, self._config.grpc_connect_attempts)
                backoff_s = self._config.grpc_retry_backoff_seconds
                self._grpc_target = (host, port)
                for attempt in range(1, attempts + 1):
                    reason = await self._connect_grpc_kernel_once(host, port)
                    if reason is None:
                        logger.info(
                            "CXP gRPC kernel connected at %s:%d for intent "
                            "dispatch (attempt %d/%d)",
                            host,
                            port,
                            attempt,
                            attempts,
                        )
                        break
                    if attempt < attempts:
                        logger.warning(
                            "CXP gRPC kernel at %s:%d unreachable "
                            "(attempt %d/%d): %s; retrying",
                            host,
                            port,
                            attempt,
                            attempts,
                            reason,
                        )
                        await asyncio.sleep(backoff_s)
                    else:
                        logger.warning(
                            "CXP gRPC kernel at %s:%d unreachable after "
                            "%d attempt(s) (%s); falling back to "
                            "in-process pipeline",
                            host,
                            port,
                            attempts,
                            reason,
                        )

        # Fallback: use PipelineKernelClient (in-process) when gRPC is unavailable
        if self._kernel_client is None:
            try:
                from orchestrata_ultimate.orchestrator.pipeline_kernel import (
                    PipelineKernelClient,
                )

                # Inject broadcast so pipeline stage events are pushed straight
                # to subscribed UI connections (in-process, no Redis required).
                pipeline_client = PipelineKernelClient(
                    pubsub=self._pubsub, broadcast=self.broadcast
                )
                await pipeline_client.connect()
                # Orchestrator-only: the in-process pipeline cannot round-trip
                # a conversational Concierge reply, so chat (cxp.send) stays
                # kernel-less here and reports the explicit assistant-
                # unavailable notice (SPEC-032 G243) instead of a fabricated
                # reply.
                self._rpc.set_pipeline_client(pipeline_client)
                self._kernel_client = pipeline_client
                self._fallback_client = pipeline_client
                self._kernel_mode = "in-process"
                self._kernel_endpoint = None
                logger.info("PipelineKernelClient connected as fallback (in-process)")
            except Exception:
                self._kernel_mode = "unavailable"
                self._kernel_endpoint = None
                logger.warning("PipelineKernelClient fallback unavailable")
            self._rpc.set_kernel_status(self._kernel_mode, self._kernel_endpoint)
        self._init_kernel_reprobe()

    def _init_pipeline_forwarder(self) -> None:
        """Forward kernel pipeline events (PipelineEventBus) to UI subscribers
        on ``pipeline.<thread_id>``. Backed by the shared Redis stream so it
        works whether the kernel is in-process or behind gRPC. Skipped when
        Redis is unavailable (ring buffer never connected).
        """
        if self._ring_buffer.client is not None:
            self._pipeline_task = asyncio.create_task(self._consume_pipeline_events())
            logger.info("Pipeline event forwarder started")

    def _init_kernel_events_forwarder(self) -> None:
        """Forward kernel lifecycle events (``OrchestrationKernel.emit_event``,
        e.g. ``kernel.boot.*``/``kernel.ready``/``kernel.stopping``/
        ``kernel.stopped``) to UI subscribers on ``kernel.events``. Backed by
        the shared Redis stream so it works whether the kernel is in-process
        or behind gRPC. Skipped when Redis is unavailable (ring buffer never
        connected) -- same gate as ``_init_pipeline_forwarder``.
        """
        if self._ring_buffer.client is not None:
            self._kernel_events_task = asyncio.create_task(
                self._consume_kernel_events()
            )
            logger.info("Kernel event forwarder started")

    def _init_config_service(self) -> None:
        """Wire the SPEC-014 US-014-005 runtime-config service: a
        ``RuntimeConfigStore`` over its OWN Redis connection (NOT env
        config -- see runtime_config.py), registered on the RPC dispatcher,
        plus a ``config:events`` forwarder task so ``config.watch``
        subscribers see live ``config.changed`` broadcasts. Best-effort
        (non-fatal): degrades to no runtime-config service when Redis is
        unavailable, matching every other optional-infra wiring step here.
        """
        try:
            from orchestrata_ultimate.core.redis_factory import create_async_redis
            from orchestrata_ultimate.gateway.runtime_config import RuntimeConfigStore

            environ = (
                self._config.environ if self._config.environ is not None else os.environ
            )
            store = RuntimeConfigStore(create_async_redis(env=environ))
            self._rpc.set_runtime_config_store(store)
            self._config_store = store
            self._config_task = asyncio.create_task(self._consume_config_events())
            logger.info("Runtime config service wired")
        except Exception:
            logger.warning("Runtime config service unavailable")

    async def _start_websocket_server(self) -> None:
        """Start the WebSocket server."""
        ws_logger = logging.getLogger("websockets.server")
        if _HANDSHAKE_NOISE_FILTER not in ws_logger.filters:
            ws_logger.addFilter(_HANDSHAKE_NOISE_FILTER)
        self._server = await websockets.serve(
            self._handle_connection,
            self._config.host,
            self._config.port,
            # SPEC-014 FR-014: ping_interval=None disables the websockets
            # library's own built-in keepalive on served connections. That
            # mechanism sends one ping per ping_interval and fails the
            # connection on a SINGLE unanswered ping_timeout -- it has no
            # concept of "N consecutive missed pongs" and its interval/
            # timeout aren't configurable per-gateway-instance (REQ-014-110).
            # _heartbeat_loop (started in start()) is this server's one and
            # only pinger, built on GatewayConfig.heartbeat_interval /
            # heartbeat_max_misses instead of fighting the library's
            # parallel schedule.
            ping_interval=None,
        )
        logger.info(
            "Gateway listening on ws://%s:%d",
            self._config.host,
            self._config.port,
        )

    def _init_heartbeat_monitor(self) -> None:
        """Start the SPEC-014 FR-014 server-initiated heartbeat ping loop."""
        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
        logger.info(
            "Heartbeat monitor started (interval=%ss, max_misses=%d)",
            self._config.heartbeat_interval,
            self._config.heartbeat_max_misses,
        )

    def _start_http_surface(self) -> None:
        """Start the HTTP surface (health/readiness/metrics) alongside the
        WS server, if enabled in config.
        """
        if self._config.health_enabled:
            from orchestrata_ultimate.gateway.http_surface import (
                build_gateway_http_app,
                serve_http,
            )

            async def _gateway_ready() -> bool:
                return self._server is not None

            http_app = build_gateway_http_app(readiness_probe=_gateway_ready)
            # http_bind_host, not host: /metrics must be reachable by the
            # Prometheus container on incusbr0, while the WS API stays on
            # whatever `host` says (127.0.0.1 in prod).
            bind = self._config.http_bind_host
            self._http_task = asyncio.create_task(
                serve_http(http_app, bind, self._config.http_port)
            )
            logger.info(
                "Gateway HTTP surface on http://%s:%d (/health /ready /metrics)",
                bind,
                self._config.http_port,
            )

    @traces(spec="014", story="US-011", reqs=["REQ-014-166"])
    async def start(self) -> None:
        """Start gateway (tolerates missing Redis/graph)."""
        await self._init_redis()
        await self._init_graph_audit()
        # Also connect the graph repository for knowledge queries
        if GRAPH_DRIVER_AVAILABLE and create_async_graph_repository is not None:
            graph_repository = self._init_graph_repository()
            # Wire test runner with broadcast capability
            self._init_test_runner(graph_repository)

        self._register_memory_engine()
        self._init_policy_engine()
        await self._init_kernel()
        self._init_pipeline_forwarder()
        self._init_kernel_events_forwarder()
        self._init_config_service()
        await self._start_websocket_server()
        self._init_heartbeat_monitor()
        self._start_http_surface()

    @staticmethod
    async def _cancel_task(task: Any) -> None:
        """Cancel a background task and wait for it to finish, tolerating
        both the expected ``CancelledError`` and any other exception raised
        during shutdown.
        """
        if task is None:
            return
        task.cancel()
        try:
            await task
        except asyncio.CancelledError, Exception:
            pass

    @traces(spec="014", story="US-011", reqs=["REQ-014-170"])
    async def stop(self) -> None:
        """Gracefully shut down the gateway."""
        if self._server:
            self._server.close()
            await self._server.wait_closed()
        # Stop the heartbeat ping loop -- no more connections will arrive
        # once the server socket is closed above, so this can stop first.
        await self._cancel_task(getattr(self, "_heartbeat_task", None))
        await self._ring_buffer.close()
        await self._audit.close()
        # Close the graph repository used by knowledge handlers
        try:
            graph_repository = getattr(self, "_graph_repository", None)
            if graph_repository:
                await graph_repository.close()
        except Exception:
            logger.warning("Failed to close graph repository")
        # Stop the kernel re-probe first: one that is mid-attempt is cancelled
        # and cannot wire a client into a gateway that is shutting down.
        reprober = getattr(self, "_kernel_reprober", None)
        if reprober is not None:
            await reprober.stop()
        # Stop the kernel reply consumer before closing its client.
        await self._cancel_task(getattr(self, "_reply_task", None))
        # Stop the pipeline-event forwarder.
        await self._cancel_task(getattr(self, "_pipeline_task", None))
        # Stop the kernel-events forwarder.
        await self._cancel_task(getattr(self, "_kernel_events_task", None))
        # Stop the runtime-config forwarder and close its store (SPEC-014).
        await self._cancel_task(getattr(self, "_config_task", None))
        config_store = getattr(self, "_config_store", None)
        if config_store is not None:
            try:
                await config_store.close()
            except Exception:
                logger.warning("Failed to close runtime config store")
        # Stop the HTTP surface (health/readiness/metrics).
        await self._cancel_task(getattr(self, "_http_task", None))
        # Close the CXP kernel client
        try:
            kernel_client = getattr(self, "_kernel_client", None)
            if kernel_client is not None:
                # The out-of-process CXP client closes with disconnect(); the
                # in-process PipelineKernelClient exposes close() instead.
                # Calling disconnect() unconditionally raised AttributeError
                # straight into the blanket handler below, so on the fallback
                # path -- the one the deployed gateway was actually running --
                # the client was never closed at all, silently.
                closer = getattr(kernel_client, "disconnect", None) or getattr(
                    kernel_client, "close", None
                )
                if closer is None:
                    logger.warning(
                        "Kernel client %s exposes neither disconnect() nor "
                        "close(); cannot close it",
                        type(kernel_client).__name__,
                    )
                else:
                    await closer()
        except Exception as exc:
            # Named, with the exception: the previous message said only
            # "Failed to close CXP kernel client", which is why the
            # AttributeError above went unnoticed.
            logger.warning(
                "Failed to close kernel client %s: %s",
                type(getattr(self, "_kernel_client", None)).__name__,
                exc,
            )
        # A gateway that switched to grpc still owns the in-process client it
        # ran before the switch (retained so in-flight pipelines finished).
        fallback = getattr(self, "_fallback_client", None)
        if fallback is not None and fallback is not getattr(
            self, "_kernel_client", None
        ):
            try:
                await fallback.close()
            except Exception as exc:
                logger.warning("Failed to close fallback pipeline client: %s", exc)
        # Stop the shared memory engine (closes its Redis/graph stores). Safe
        # when never connected — stop() just clears state and closes idle stores.
        engine = getattr(self, "_memory_engine", None)
        if engine is not None:
            try:
                await engine.stop()
            except Exception:
                logger.warning("Failed to stop memory engine")

    @traces(
        spec="014",
        story="US-007",
        reqs=["REQ-014-106", "REQ-014-108", "REQ-014-110"],
    )
    async def _heartbeat_loop(self) -> None:
        """Server-initiated WebSocket ping loop (SPEC-014 FR-014).

        Every ``heartbeat_interval`` seconds (REQ-014-106, REQ-014-110 --
        both configurable per gateway instance via ``GatewayConfig``), sends
        a WebSocket ping to each live connection and checks whether the
        PREVIOUS tick's ping was ever ponged. A connection whose pong is
        still outstanding after ``heartbeat_max_misses`` consecutive ticks
        is force-disconnected (REQ-014-108) via
        ``_disconnect_timed_out_connection``.

        This is the gateway's ONLY pinger -- the websockets library's own
        built-in keepalive is disabled in ``_start_websocket_server`` (see
        the comment there) because it has no consecutive-miss concept.

        Runs for the life of the server; started in ``start()``, cancelled
        in ``stop()``.
        """
        misses: dict[str, int] = {}
        pending: dict[str, Any] = {}
        while True:
            await asyncio.sleep(self._config.heartbeat_interval)
            live_ids = set(self._ws_map)
            for stale_id in set(misses) | set(pending):
                if stale_id not in live_ids:
                    # Connection closed on its own between ticks (client
                    # disconnect, auth failure, etc.) -- nothing to do.
                    misses.pop(stale_id, None)
                    pending.pop(stale_id, None)
            for conn_id, ws in list(self._ws_map.items()):
                waiter = pending.pop(conn_id, None)
                if waiter is not None:
                    misses[conn_id] = 0 if waiter.done() else misses.get(conn_id, 0) + 1
                if misses.get(conn_id, 0) >= self._config.heartbeat_max_misses:
                    misses.pop(conn_id, None)
                    # Remove from ws_map immediately: this connection is
                    # already considered gone (no further pings, pushes, or
                    # broadcasts to it), independent of how long the actual
                    # socket teardown below takes.
                    self._ws_map.pop(conn_id, None)
                    asyncio.create_task(
                        self._disconnect_timed_out_connection(conn_id, ws)
                    )
                    continue
                try:
                    pending[conn_id] = await ws.ping()
                except websockets.ConnectionClosed:
                    misses.pop(conn_id, None)

    @traces(spec="014", story="US-007", reqs=["REQ-014-114"])
    async def _disconnect_timed_out_connection(self, conn_id: str, ws: Any) -> None:
        """Force-close a connection that missed ``heartbeat_max_misses``
        consecutive pongs, publishing ``gateway.presence`` (reason=timeout).

        Publishes first: the event fires the instant the miss threshold is
        reached, independent of how long the actual socket teardown takes --
        a genuinely unresponsive client may never acknowledge the closing
        handshake at all. The close attempt itself is bounded by
        ``heartbeat_interval`` rather than the library's fixed 10s
        ``close_timeout`` default: a peer that already missed
        ``heartbeat_max_misses`` pings has had ample chance to respond, so
        there's no reason to wait a further fixed 10s for a close ack it is
        unlikely to send -- a stubborn connection is force-aborted instead
        once that bound elapses.

        Args:
            conn_id: The timed-out connection's identifier.
            ws: Its websocket connection object (already removed from
                ``self._ws_map`` by the caller).
        """
        await self.broadcast(
            "gateway.presence",
            {
                "connection_id": conn_id,
                "event": "disconnected",
                "reason": "timeout",
            },
        )
        try:
            await asyncio.wait_for(
                ws.close(code=1008, reason="heartbeat timeout"),
                timeout=self._config.heartbeat_interval,
            )
        except Exception:
            transport = getattr(ws, "transport", None)
            if transport is not None:
                transport.abort()

    async def _consume_kernel_replies(self) -> None:
        """Background task: pump volk replies from the kernel to UI subscribers
        on ``cxp.traffic`` until cancelled. Closes the intent→reply loop for the
        out-of-process kernel.

        When the gRPC server does not implement streaming ``receive_packet``,
        failures are expected and the consumer backs off exponentially before
        giving up (the PipelineKernelClient fallback handles replies in-process)."""
        client = self._kernel_client
        if client is None:
            return
        consecutive_failures = 0
        while True:
            try:
                await _pump_kernel_replies_once(
                    client,
                    self.broadcast,
                    agent_id=self._reply_consumer_id,
                    timeout_seconds=self._config.kernel_reply_timeout_s,
                )
                consecutive_failures = 0
            except asyncio.CancelledError:
                break
            except Exception as exc:
                consecutive_failures += 1
                if consecutive_failures == 1:
                    logger.warning(
                        "Kernel reply consumer failed (%s); retrying with backoff",
                        exc,
                    )
                elif consecutive_failures >= 10:
                    logger.warning(
                        "Kernel reply consumer gave up after %d consecutive failures",
                        consecutive_failures,
                    )
                    break
                backoff = min(
                    self._config.kernel_reply_backoff_base_s
                    * (2**consecutive_failures),
                    60,
                )
                await asyncio.sleep(backoff)

    async def _consume_pipeline_events(self) -> None:
        """Background task: forward kernel pipeline events to UI subscribers on
        ``pipeline.<thread_id>`` until cancelled. Tails the shared Redis stream
        written by ``PipelineEventBus`` so stage transitions appear live in the
        CXPFeed."""
        redis_client = self._ring_buffer.client
        if redis_client is None:
            return
        # Start from new events only; reconnecting clients replay durably via the
        # per-thread ring-buffer stream, not this live tail.
        last_id = "$"
        while True:
            try:
                last_id, _ = await _pump_pipeline_events_once(
                    redis_client,
                    self.broadcast,
                    last_id,
                    stream_key=self._pipeline_stream_key,
                )
            except asyncio.CancelledError:
                break
            except Exception:
                logger.warning("Pipeline event forwarder error; continuing")
                await asyncio.sleep(1)

    @traces(spec="014", story="US-012", reqs=["REQ-014-192"])
    async def _consume_kernel_events(self) -> None:
        """Background task: forward kernel lifecycle events to UI subscribers
        on ``kernel.events`` until cancelled. Tails the shared Redis stream
        written by ``EventBus._publish_to_redis`` so vizint-ui's
        NotificationCenter shows boot/ready/shutdown transitions live."""
        redis_client = self._ring_buffer.client
        if redis_client is None:
            return
        # Start from new events only; the per-thread pipeline ring-buffer
        # streams handle durable replay for reconnecting clients, not this
        # live tail.
        last_id = "$"
        while True:
            try:
                last_id = await _pump_kernel_events_once(
                    redis_client,
                    last_id,
                    self.broadcast,
                    stream_key=self._kernel_events_stream_key,
                )
            except asyncio.CancelledError:
                break
            except Exception:
                logger.warning("Kernel event forwarder error; continuing")
                await asyncio.sleep(1)

    async def _consume_config_events(self) -> None:
        """Background task: forward runtime-config changes to UI subscribers
        on ``config.changed`` until cancelled (SPEC-014 REQ-014-081). Clone
        of ``_consume_pipeline_events`` for ``RuntimeConfigStore``'s
        ``config:events`` stream -- delivers via ``MessageProcessor.broadcast``,
        never ``PubSubManager.publish`` (D7)."""
        store = self._config_store
        if store is None:
            return
        while True:
            try:
                await _pump_config_events_once(store, self.broadcast)
            except asyncio.CancelledError:
                break
            except Exception:
                logger.warning("Config event forwarder error; continuing")
                await asyncio.sleep(1)

    @property
    def port(self) -> int:
        """Bound port number."""
        return self._config.port

    @property
    def processor(self) -> MessageProcessor:
        """Expose the message processor for push operations."""
        return self._processor

    # -- per-connection handler ------------------------------

    @traces(spec="014", story="US-011", reqs=["REQ-014-168"])
    async def _handle_connection(self, websocket: Any) -> None:
        """Handle a single WebSocket connection lifecycle.

        Args:
            websocket: The websockets connection object.
        """
        conn_id = str(uuid.uuid4())
        logger.info(f"[GATEWAY] New connection request: {conn_id}")
        new_correlation_id()
        try:
            remote = getattr(websocket, "remote_address", ("unknown", 0))
            client_ip = remote[0] if remote else "unknown"
            user_id = await authenticate(
                websocket,
                conn_id,
                self._config,
                self._auth,
                self._connections,
                self._ws_map,
                self._handle_jsonrpc,
            )
            if not user_id:
                logger.info(
                    "WebSocket auth failed",
                    extra={
                        "client_ip": client_ip,
                        "auth_result": "failed",
                    },
                )
                return
            logger.info(
                "WebSocket connected",
                extra={
                    "client_ip": client_ip,
                    "auth_result": "ok",
                    "scope": "read_write",
                },
            )
            await self._connections.register(conn_id, user_id)
            self._ws_map[conn_id] = websocket
            await self._audit.log_event(
                "connection.open",
                conn_id,
                {"user_id": user_id},
            )
            async for raw in websocket:
                await self._handle_message(websocket, conn_id, raw)
        except websockets.ConnectionClosed:
            logger.debug("Connection %s closed", conn_id)
        finally:
            self._ws_map.pop(conn_id, None)
            await self._pubsub.unsubscribe_all(conn_id)
            await self._connections.unregister(conn_id)
            await self._audit.log_event("connection.close", conn_id)

    # -- message routing -------------------------------------

    @traces(spec="014", story="US-011", reqs=["REQ-014-176"])
    @traces(spec="014", story="US-007", reqs=["REQ-014-109"])
    async def _handle_message(
        self,
        websocket: Any,
        conn_id: str,
        raw: str,
    ) -> None:
        """Route an inbound message -- JSON-RPC 2.0 or internal.

        Refreshes ``last_activity`` (``ConnectionManager.update_heartbeat``,
        REQ-014-109) for EVERY inbound frame first, before parsing or
        routing -- this is the single choke point every message from a
        connected client passes through (``_handle_connection``'s
        ``async for raw in websocket`` calls only this), so it covers both
        the JSON-RPC and legacy WSMessage paths, and even malformed JSON --
        a client that sends garbage is still observably alive.

        Args:
            websocket: The websockets connection object.
            conn_id: Connection identifier.
            raw: Raw message string from the client.
        """
        await self._connections.update_heartbeat(conn_id)
        logger.info(f"[GATEWAY] RAW MESSAGE from {conn_id}: {raw[:500]}")
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            await websocket.send(
                _jsonrpc_error(None, JSONRPC_PARSE_ERROR, "Parse error")
            )
            return
        if "jsonrpc" in data:
            await self._handle_jsonrpc(websocket, conn_id, data)
        else:
            await self._processor.process(conn_id, raw)

    @traces(spec="014", story="US-011", reqs=["REQ-014-177", "REQ-014-180"])
    async def _handle_jsonrpc(
        self,
        websocket: Any,
        conn_id: str,
        data: dict[str, Any],
    ) -> None:
        """Handle a JSON-RPC 2.0 request.

        Args:
            websocket: The websockets connection object.
            conn_id: Connection identifier.
            data: Parsed JSON-RPC message dict.
        """
        method = data.get("method", "")
        params = data.get("params", {})
        msg_id = data.get("id")
        if not method:
            await websocket.send(
                _jsonrpc_error(msg_id, JSONRPC_INVALID_REQUEST, "Missing method")
            )
            return
        t0 = time.monotonic()
        try:
            result = await self._rpc.dispatch(method, conn_id, params)
            duration_ms = (time.monotonic() - t0) * 1000
            logger.info(
                "RPC call",
                extra={
                    "method": method,
                    "params": str(params)[:200],
                    "result": "ok",
                    "duration_ms": round(duration_ms, 2),
                },
            )
            await websocket.send(_jsonrpc_result(msg_id, result))
        except RPCError as exc:
            await websocket.send(_jsonrpc_error(msg_id, exc.rpc_code, str(exc)))
        except Exception:
            cid = correlation_id.get("")
            logger.exception(
                "Unhandled RPC error [correlation_id=%s]",
                cid,
            )
            await websocket.send(
                _jsonrpc_error(
                    msg_id,
                    JSONRPC_INTERNAL_ERROR,
                    "Internal server error",
                    {"correlation_id": cid},
                )
            )

    # -- push ------------------------------------------------

    @traces(spec="014", story="US-011", reqs=["REQ-014-174"])
    async def push_to_client(
        self,
        connection_id: str,
        method: str,
        params: dict[str, Any],
    ) -> bool:
        """Send a JSON-RPC notification to a specific client.

        Args:
            connection_id: Target client connection ID.
            method: Notification method name.
            params: Notification parameters.

        Returns:
            True if delivered, False otherwise.
        """
        return await self._processor.push_to_client(connection_id, method, params)

    @traces(spec="014", story="US-012", reqs=["REQ-014-131", "REQ-014-193"])
    async def broadcast(
        self,
        topic: str,
        event_payload: dict[str, Any],
    ) -> int:
        """Broadcast a JSON-RPC notification to all subscribers.

        Args:
            topic: Event topic name.
            event_payload: Event data.

        Returns:
            Number of clients delivered to.
        """
        return await self._processor.broadcast(topic, event_payload)


# Class alias for orchestrata-ultimate naming
class VizIntWebUI(WebSocketGateway):
    """WebUI alias for WebSocketGateway."""

    pass