orchestrata_ultimate.kernel.volk_status_bridge

Reference › orchestrata_ultimate › kernel › volk_status_bridge

A small adapter that turns a volk status change into a kernel event. VolkStatusEventBridge is handed to the volk registry (agents.volk_registry) as its on_status_change callback, which the registry calls synchronously and outside its lock. The bridge schedules a volk.status_changed event carrying the alias, the old status and the new status through an async emit callable, which cli_main supplies at boot. The module serves SPEC-001 requirements REQ-001-014, REQ-001-009 and REQ-001-022.

The callback never raises into the registry. If building the event fails, on_status_change logs a warning and returns. _schedule then runs the coroutine without blocking, in three tiers: create_task on the running loop of the current thread; otherwise run_coroutine_threadsafe onto the loop given to bind_loop, which is the case when the callback fires on the daemon thread of the heartbeat pump; otherwise the coroutine is closed and dropped. Task results are logged by _log_task_result. Events end up on the EventBus.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkStatusEventBridge

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-014', 'REQ-001-009', 'REQ-001-022']) · Line 33

Adapts VolkRegistry's sync on_status_change callback onto the async kernel-event emit seam.

Source

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

"""SPEC-001 volk-comms-heartbeat — the status-change -> kernel-event bridge (closes G3d).

Registry-side status transitions (``VolkRegistry``'s ``on_status_change``
seam, decision 3 of ``plans/volk-comms-heartbeat-design.md`` §4.1) were only
ever consumed by the roster itself -- nothing pushed them onward, so a
kernel-side ACTIVE<->INACTIVE flip never reached the gateway/UI in real time
(REQ-001-014's subscription contract was unwired). ``VolkStatusEventBridge``
is a tiny observer adapter: its ``on_status_change`` method IS the callback
``VolkRegistry(on_status_change=...)`` fires, and it re-emits the transition
as a ``volk.status_changed`` kernel event through the already-tested
kernel-events -> gateway broadcast seam (``gateway/server.py``
``_init_kernel_events_forwarder`` / ``_pump_kernel_events_once``) -- no new
event framework, reusing the existing ``OrchestrationKernel.emit_event`` path
(decision 5).
"""

from __future__ import annotations

import asyncio
import concurrent.futures
import logging
from collections.abc import Awaitable, Callable
from typing import Any

from orchestrata_ultimate.agents.volk_registry import VolkStatus
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer

logger = logging.getLogger(__name__)


@traces(spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-009", "REQ-001-022"])
class VolkStatusEventBridge:
    """Adapts ``VolkRegistry``'s sync ``on_status_change`` callback onto the
    async kernel-event emit seam.

    Composed of an *emit* callable -- the production construction site
    (``kernel/cli_main.py`` boot) binds it to
    ``functools.partial(_emit_kernel_event, kernel)``, the already-tested
    helper proven end-to-end by ``tests/kernel/test_cli_main_kernel_events.py``
    -- so this module has no direct dependency on ``OrchestrationKernel``
    itself, only on the narrow ``(event_type, message, **payload) ->
    Awaitable`` shape ``_emit_kernel_event`` already exposes.
    """

    def __init__(
        self,
        emit: Callable[..., Awaitable[Any]],
        tracer: Tracer | None = None,
    ) -> None:
        """Bind this bridge to *emit*.

        Args:
            emit: Async callable ``(event_type: str, message: str,
                **payload) -> Awaitable[None]`` -- the kernel-event emission
                seam (``cli_main._emit_kernel_event`` partially bound to a
                live ``kernel``, in production).
            tracer: Optional span/log emitter; defaults to the process-wide
                Singleton (:meth:`Tracer.default`).
        """
        self._emit = emit
        self._tracer = tracer if tracer is not None else Tracer.default()
        self._bound_loop: asyncio.AbstractEventLoop | None = None

    @traces(
        spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-009", "REQ-001-022"]
    )
    def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None:
        """Bind *loop* as the off-thread scheduling fallback for
        :meth:`on_status_change`.

        ``VolkRegistry`` fires its ``on_status_change`` callback
        synchronously from WHATEVER thread drives the transition -- in
        production that is ``HeartbeatPump``'s daemon ``threading.Thread``
        (``kernel/heartbeat_pump.py``'s ``_run``) beating an idle volk back
        to ACTIVE, not the async kernel's own event loop. Without a bound
        loop, ``_schedule``'s ``asyncio.get_running_loop()`` call raises
        ``RuntimeError`` on that thread and the event is silently dropped
        every time -- INACTIVE->ACTIVE revivals never reach the UI. The
        production construction site
        (``kernel/cli_main.py::_boot_intent_pipeline``, itself a coroutine)
        binds its own running loop here immediately after constructing the
        bridge.
        """
        self._bound_loop = loop

    @traces(
        spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-009", "REQ-001-022"]
    )
    def on_status_change(self, alias: str, old: VolkStatus, new: VolkStatus) -> None:
        """``VolkRegistry``'s ``on_status_change`` seam (REQ-001-014):
        schedule a ``volk.status_changed`` kernel event for this transition.

        Best-effort (mirrors ``VolkRegistry._fire_status_change``,
        REQ-001-022/REQ-001-017): a failure building or emitting the event
        is only logged -- this callback must NEVER raise into the registry,
        which invokes it synchronously, outside its lock (REQ-001-014's
        contract, decision 3 of the design).
        """
        with self._tracer.span("VolkStatusEventBridge.on_status_change"):
            try:
                coro = self._emit(
                    "volk.status_changed",
                    f"Volk {alias} transitioned {old.value} -> {new.value}",
                    alias=alias,
                    old_status=old.value,
                    new_status=new.value,
                )
            except Exception:
                logger.warning(
                    "VolkStatusEventBridge: emit() raised building the "
                    "volk.status_changed event for %s",
                    alias,
                    exc_info=True,
                )
                return
            self._schedule(coro, alias)

    @traces(
        spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-009", "REQ-001-022"]
    )
    def _schedule(self, coro: Any, alias: str) -> None:
        """Fire-and-forget *coro*, never blocking or raising into the caller.

        Three tiers, in order:

        1. A loop is running on the CURRENT thread (this callback fired from
           async code, or the caller's thread happens to host one) ->
           ``loop.create_task`` (unchanged fast path).
        2. No loop on the current thread, but :meth:`bind_loop` bound one
           that is still running (the production case: this callback fires
           on ``HeartbeatPump``'s daemon thread) ->
           ``asyncio.run_coroutine_threadsafe`` delivers *coro* onto that
           loop instead of dropping it.
        3. Neither -- *coro* is closed immediately, the same "zero behavior
           change" outcome as before (mirrors
           ``kernel.audit_bootstrap.fire_and_forget_audit``'s final
           fallback).
        """
        if not asyncio.iscoroutine(coro):
            return
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None
        if loop is not None:
            task = loop.create_task(coro)
            task.add_done_callback(lambda t: self._log_task_result(t, alias))
            return
        bound_loop = self._bound_loop
        if bound_loop is not None and bound_loop.is_running():
            future = asyncio.run_coroutine_threadsafe(coro, bound_loop)
            future.add_done_callback(lambda f: self._log_task_result(f, alias))
            return
        coro.close()

    @traces(
        spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-009", "REQ-001-022"]
    )
    def _log_task_result(
        self,
        task: asyncio.Task[Any] | concurrent.futures.Future[Any],
        alias: str,
    ) -> None:
        if task.cancelled():
            return
        exc = task.exception()
        if exc is not None:
            logger.warning(
                "VolkStatusEventBridge: volk.status_changed emission failed "
                "for %s: %s",
                alias,
                exc,
            )