orchestrata_ultimate.core.protocols.volk_health

Reference › orchestrata_ultimate › core › protocols › volk_health

The narrow health seam of SPEC-001. VolkHealthProbe has two methods: record_heartbeat(alias), which returns False for an unknown volk, and get_volk_health(alias), which returns the volk's health if it is registered. Routing and tool-call code, namely kernel.heartbeat_pump and kernel.pipeline and the agents' tool executor, use it instead of the concrete volk registry. The registry satisfies it structurally, and this layer-0 module never imports the registry. Its methods carry @traces annotations for requirements REQ-001-018 and REQ-001-015.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkHealthProbe (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='001', story='US-001', reqs=['REQ-001-018', 'REQ-001-015']) · Line 20

Structural health-seam contract satisfied by the Volk registry.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/protocols/volk_health.py

"""Volk health probe protocol — the narrow SPEC-001 health seam.

``VolkHealthProbe`` is the minimal structural contract that routing and
tool-call seams (e.g. ``kernel.heartbeat_pump.HeartbeatPump``,
``agents.tool_registry.ToolExecutor``) compose against instead of depending
on the concrete Volk registry surface. The concrete registry (Layer 3)
satisfies this Protocol structurally — no inheritance required, and this
Layer-0 module never imports it.
"""

from __future__ import annotations

from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces


@runtime_checkable
@traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-015"])
class VolkHealthProbe(Protocol):
    """Structural health-seam contract satisfied by the Volk registry."""

    # record_heartbeat: record a heartbeat for a Volk; False if unknown.
    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def record_heartbeat(self, alias: str) -> bool: ...

    # get_volk_health: current health status of a Volk, if registered.
    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def get_volk_health(self, alias: str) -> Any | None: ...

    # is_registered: whether the given alias is currently registered.
    @traces(spec="001", story="US-001", reqs=["REQ-001-015"])
    def is_registered(self, alias: str) -> bool: ...