orchestrata_ultimate.gateway.kernel_reprobe

Reference › orchestrata_ultimate › gateway › kernel_reprobe

kernel_reprobe upgrades the gateway from its in-process fallback to the real kernel once the kernel becomes reachable (SPEC-014 US-014-011, REQ-014-166 and REQ-014-170). At boot server tries the cxpt direct transport to the kernel in a bounded way and falls back to PipelineKernelClient if that fails; this module keeps retrying in the background so a kernel that binds its port late is picked up without a gateway restart.

KernelConnectAttempt is a protocol for one bounded attempt that returns True when the kernel is wired. KernelReprober takes that callable, an interval, the endpoint (used only in the log line) and an optional tracer. start creates a named asyncio task and is idempotent; it returns None when the interval is zero or negative, which disables the loop (the GATEWAY_GRPC_REPROBE_INTERVAL_S setting in config). The loop sleeps for the interval, counts the attempt, runs it inside a gateway.kernel_reprobe.attempt span, treats an exception as a failed attempt, and ends after the first success, setting switched. stop cancels and awaits the task and is safe to call twice. The properties enabled, attempts, switched and task expose its state. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

KernelConnectAttempt (protocol)

Bases: Protocol · Decorators: traces(spec='SPEC-014', story='US-014-005', reqs=['REQ-014-071', 'REQ-014-075']), runtime_checkable · Line 31

Try once to wire the real kernel; True when wired.

KernelReprober

Decorators: traces(spec='SPEC-014', story='US-014-011', reqs=['REQ-014-166', 'REQ-014-170']) · Line 38

Periodically retry a kernel connect attempt until it succeeds.

Source

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

"""Bounded background re-probe: upgrade the in-process fallback to the real kernel.

``WebSocketGateway._init_kernel`` probes the CXP kernel a few times at boot and,
if it is unreachable, permanently installs the in-process
``PipelineKernelClient``. The kernel may bind its port seconds *after* the
gateway booted, so :class:`KernelReprober` keeps trying — one bounded attempt
per interval — and ends on the first success (contract:
``plans/contracts/gateway-kernel-reprobe.contract.json``).
"""

from __future__ import annotations

import asyncio
import logging
import time
from typing import Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer

logger = logging.getLogger(__name__)

_SWITCH_LINE = (
    "CXP gRPC kernel at %s reachable after %d re-probe(s) over %.1fs; "
    "switched intent dispatch from in-process fallback to grpc"
)


@traces(spec="SPEC-014", story="US-014-005", reqs=["REQ-014-071", "REQ-014-075"])
@runtime_checkable
class KernelConnectAttempt(Protocol):
    """Try once to wire the real kernel; ``True`` when wired."""

    async def __call__(self) -> bool: ...


@traces(spec="SPEC-014", story="US-014-011", reqs=["REQ-014-166", "REQ-014-170"])
class KernelReprober:
    """Periodically retry a kernel connect attempt until it succeeds."""

    def __init__(
        self,
        attempt: KernelConnectAttempt,
        interval_seconds: float,
        endpoint: str,
        tracer: Tracer | None = None,
    ) -> None:
        """Wire KernelReprober's composed collaborators.

        Args:
            attempt: One bounded connect attempt; ``True`` when the kernel is wired.
            interval_seconds: Sleep between attempts; ``<= 0`` disables the loop.
            endpoint: ``host:port`` used only in the switch log line.
            tracer: Span source; defaults to the process-wide ``Tracer``.
        """
        self._attempt = attempt
        self._interval_seconds = interval_seconds
        self._endpoint = endpoint
        self._tracer = tracer if tracer is not None else Tracer.default()
        self._task: asyncio.Task[None] | None = None
        self._attempts = 0
        self._switched = False

    @property
    def enabled(self) -> bool:
        """True when the interval is positive (0 disables the re-probe)."""
        return self._interval_seconds > 0

    @property
    def attempts(self) -> int:
        """Re-probes made so far."""
        return self._attempts

    @property
    def switched(self) -> bool:
        """True once an attempt succeeded."""
        return self._switched

    @property
    def task(self) -> asyncio.Task[None] | None:
        """The running loop task, or ``None`` before start / after stop."""
        return self._task

    @traces(reqs=["REQ-014-166"])
    def start(self) -> asyncio.Task[None] | None:
        """Create the loop task; ``None`` when disabled.

        Idempotent: a second call returns the already-running task. Must be
        called from inside the running event loop.
        """
        if not self.enabled:
            return None
        if self._task is not None:
            return self._task
        self._task = asyncio.create_task(self._run(), name="gateway-kernel-reprobe")
        return self._task

    @traces(reqs=["REQ-014-170"])
    async def stop(self) -> None:
        """Cancel and await the loop task; safe to call twice."""
        task = self._task
        if task is None:
            return
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.debug("kernel re-probe task raised during stop", exc_info=True)
        self._task = None

    @traces(reqs=["REQ-014-071", "REQ-014-166"])
    async def _run(self) -> None:
        started = time.monotonic()
        while True:
            await asyncio.sleep(self._interval_seconds)
            self._attempts += 1
            try:
                with self._tracer.span("gateway.kernel_reprobe.attempt"):
                    ok = await self._attempt()
            except Exception as exc:
                logger.debug("kernel re-probe attempt raised: %s", exc)
                ok = False
            if ok:
                self._switched = True
                logger.info(
                    _SWITCH_LINE,
                    self._endpoint,
                    self._attempts,
                    time.monotonic() - started,
                )
                return
            logger.debug(
                "kernel re-probe %d at %s failed; next in %.1fs",
                self._attempts,
                self._endpoint,
                self._interval_seconds,
            )