orchestrata_ultimate.core.protocols.decoupling

Reference › orchestrata_ultimate › core › protocols › decoupling

Protocols that break the dependency cycles between the kernel, the orchestrator and governance (phase 1 of the cycle-breaking work). The module has no dependencies on the rest of the package except the @traces decorator, so any layer can depend on a contract here instead of on another subsystem's implementation. The kernel pipeline imports it, and interfaces.decoupling keeps the historical names.

• GovernancePrologBridgeProtocol: load rule files, run Prolog queries (query, query_one, query_bool), assert and retract facts, manage session and request scopes, and clean up expired facts.
• VolkRegistryProtocol, with RegisteredVolkProtocol and PluginVolkRegistryProtocol: register, fetch, list and query volks by capability or protocol, record heartbeats, report health and count in-flight messages.
• AuditTrailProtocol: log_* methods for violations, escalations, resolutions, timeouts, policy decisions, sessions and tool calls, plus queries over the trail, a health check and a dropped-event count.
• RedisStateManagerProtocol: sessions, a cache with governance-decision caching, distributed locks, publish and subscribe, and queues.

Each method is annotated with @traces for its spec and requirement.

At a glance

Class diagram

Dependency diagram

API

Classes

GovernancePrologBridgeProtocol (protocol)

Bases: Protocol · Line 17

Protocol for Prolog governance bridge — breaks kernel/orchestrator cycle.

VolkRegistryProtocol (protocol)

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

Protocol for Volk registration and lookup — breaks kernel/orchestrator cycle.

RegisteredVolkProtocol (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-002', reqs=['REQ-015-021']) · Line 107

Structural shape of a plugin-registered Volk record (SPEC-015 projection seam), returned by :meth:PluginVolkRegistryProtocol.get_volk.

PluginVolkRegistryProtocol (protocol)

Bases: Protocol · Decorators: traces(spec='015', story='US-002', reqs=['REQ-015-017', 'REQ-015-021']) · Line 129

Protocol for the PLUGIN-side volk registry (SPEC-015 projection seam).

AuditTrailProtocol (protocol)

Bases: Protocol · Line 148

Protocol for audit trail persistence — breaks governance/orchestrator cycle.

RedisStateManagerProtocol (protocol)

Bases: Protocol · Line 223

Protocol for Redis-backed state management — breaks governance/orchestrator cycle.

Source

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

"""
Decoupling protocols for Orchestrata Ultimate (Phase 1 - Cycle Breaking).

These protocols break the kernel/orchestrator/governance dependency cycles.
This module has no intra-package dependencies.
"""

from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces

# ============================================================
# Decoupling Protocols (Phase 1 - Cycle Breaking)
# ============================================================


class GovernancePrologBridgeProtocol(Protocol):
    """Protocol for Prolog governance bridge — breaks kernel/orchestrator cycle."""

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

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-025"])
    def load_all_rules(self) -> int: ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-007"])
    def query(self, query_str: str) -> Any: ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-007"])
    def query_one(self, query_str: str) -> dict[str, Any] | None: ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-007"])
    def query_bool(self, query_str: str) -> bool: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def assert_fact(
        self,
        predicate: str,
        args: list[Any],
        scope: Any = None,
        session_id: str | None = None,
        ttl_seconds: int | None = None,
    ) -> str: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-011"])
    def retract_fact(self, fact_id: str) -> bool: ...

    @traces(spec="007", story="FR-PGOV-010", reqs=["REQ-007-FR10-041"])
    def start_session(
        self, session_id: str, goals: list[str], context: dict[str, Any]
    ) -> None: ...

    @traces(spec="006", story="US-005", reqs=["REQ-006-FR5-001"])
    def end_session(self, session_id: str) -> int: ...

    def start_request(self, session_id: str, tool: str, intent: str) -> None: ...
    def end_request(self) -> int: ...
    def cleanup_expired_facts(self) -> int: ...
    def get_fact_count(self) -> dict[str, int]: ...


@traces(spec="001", story="US-001", reqs=["REQ-001-001"])
class VolkRegistryProtocol(Protocol):
    """Protocol for Volk registration and lookup — breaks kernel/orchestrator cycle."""

    @traces(spec="001", story="US-001", reqs=["REQ-001-008"])
    def register_volk(self, definition: Any) -> Any: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-015"])
    def get_volk(self, alias: str) -> Any | None: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-005"])
    def is_registered(self, alias: str) -> bool: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-011"])
    def list_volks(self) -> list: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-012"])
    def list_volks_by_capability(self, capability_name: str) -> list: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-013"])
    def list_volks_by_protocol(self, protocol: Any) -> list: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-021"])
    def deregister_volk(self, alias: str, graceful_timeout: float = 5.0) -> bool: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-019"])
    def query_volks(self, query_filter: Any) -> list: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def record_heartbeat(self, alias: str) -> bool: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def get_volk_health(self, alias: str) -> Any | None: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-025"])
    def increment_in_flight_messages(self, alias: str) -> bool: ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-025"])
    def decrement_in_flight_messages(self, alias: str) -> bool: ...

    @property
    def count(self) -> int: ...


@traces(spec="015", story="US-002", reqs=["REQ-015-021"])
class RegisteredVolkProtocol(Protocol):
    """Structural shape of a plugin-registered Volk record (SPEC-015
    projection seam), returned by
    :meth:`PluginVolkRegistryProtocol.get_volk`.

    Deliberately narrow — backlog [411] — to the one attribute the kernel
    pipeline's plugin-volk projection
    (``kernel/pipeline.py::_register_plugin_volks``) actually reads off the
    lookup result, replacing that projection site's bare ``Any | None``.
    ``plugin.registries.RegisteredVolk`` is the production implementer.

    Declared as a read-only ``@property`` rather than a plain ``handler:
    Any`` attribute: a bare attribute annotation makes a Protocol member
    both gettable AND settable, which the production implementer (a frozen
    dataclass) structurally fails since a frozen field has no setter.
    """

    @property
    def handler(self) -> Any: ...


@traces(spec="015", story="US-002", reqs=["REQ-015-017", "REQ-015-021"])
class PluginVolkRegistryProtocol(Protocol):
    """Protocol for the PLUGIN-side volk registry (SPEC-015 projection seam).

    Deliberately narrow — the two members the kernel pipeline's plugin-volk
    projection actually consumes. Note the semantics differ from
    ``VolkRegistryProtocol``: ``list_volks`` yields ALIAS STRINGS (not
    registration records) and ``get_volk`` returns a handler-bearing
    ``RegisteredVolkProtocol``-shaped record (``RegisteredVolkProtocol |
    None``). ``plugin.registries.VolkRegistry`` is the production
    implementer.
    """

    @traces(spec="015", story="US-002", reqs=["REQ-015-017"])
    def list_volks(self) -> list[str]: ...

    @traces(spec="015", story="US-002", reqs=["REQ-015-021"])
    def get_volk(self, alias: str) -> RegisteredVolkProtocol | None: ...


class AuditTrailProtocol(Protocol):
    """Protocol for audit trail persistence — breaks governance/orchestrator cycle."""

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def start(self) -> None: ...

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

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-020"])
    async def log_violation(self, violation: dict) -> None: ...

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-003"])
    async def log_escalation(self, result: Any) -> None: ...

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-024"])
    async def log_resolution(self, escalation_id: str, resolution: Any) -> None: ...

    async def log_timeout(self, escalation_id: str) -> None: ...

    @traces(
        spec="003",
        story="US-010",
        reqs=["REQ-003-US-010-006", "REQ-003-US-010-018"],
    )
    async def log_event(self, event: Any) -> bool: ...

    @traces(spec="007", story="US-PGOV-001", reqs=["REQ-007-US-001-011"])
    async def log_policy_decision(self, decision: Any) -> None: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-010"])
    async def log_session_start(
        self, session_id: str, volk_id: str, goals: list, context: dict
    ) -> None: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-029"])
    async def log_session_end(
        self, session_id: str, volk_id: str, outcome: str, stats: dict
    ) -> None: ...

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def log_tool_call(
        self,
        session_id: str,
        volk_id: str,
        tool: str,
        args: dict,
        result: Any,
        duration_ms: int,
    ) -> None: ...

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-024"])
    async def get_escalations(self, limit: int = 100) -> list: ...

    @traces(spec="007", story="FR-PGOV-007", reqs=["REQ-007-FR7-008"])
    async def get_violations(self, limit: int = 100) -> list: ...

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-024"])
    async def get_resolutions(self, limit: int = 100) -> list: ...

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-027"])
    async def query_audit_trail(self, filters: dict) -> list: ...

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

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-023"])
    def get_dropped_events_count(self) -> int: ...

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

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def aclose(self) -> None: ...


class RedisStateManagerProtocol(Protocol):
    """Protocol for Redis-backed state management — breaks governance/orchestrator cycle."""

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-005"])
    def create_session(self, session: Any, ttl_seconds: int = 3600) -> bool: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-015"])
    def get_session(self, session_id: str) -> Any | None: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-016"])
    def update_session(self, session_id: str, updates: dict) -> bool: ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-023"])
    def delete_session(self, session_id: str) -> bool: ...

    def cache_set(self, key: str, value: Any, ttl_seconds: float = 300) -> None: ...
    def cache_get(self, key: str) -> Any | None: ...
    def cache_delete(self, key: str) -> bool: ...

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    def cache_governance_decision(
        self, intent_hash: str, decision: dict, ttl_seconds: float = 60
    ) -> None: ...

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-005"])
    def get_cached_decision(self, intent_hash: str) -> dict | None: ...

    def acquire_lock(
        self, lock_name: str, holder_id: str, ttl_seconds: int = 30
    ) -> bool: ...
    def release_lock(self, lock_name: str, holder_id: str) -> bool: ...

    def extend_lock(
        self, lock_name: str, holder_id: str, ttl_seconds: int = 30
    ) -> bool: ...
    def publish(self, channel: str, message: dict) -> int: ...
    def subscribe(self, channel: str, callback: Any) -> None: ...

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-022"])
    def queue_push(self, queue_name: str, item: dict) -> int: ...

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-022"])
    def queue_pop(self, queue_name: str, timeout: int = 0) -> dict | None: ...

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-025"])
    def queue_length(self, queue_name: str) -> int: ...

    def health_check(self) -> bool: ...
    def close(self) -> None: ...
    async def aclose(self) -> None: ...


__all__ = [
    "GovernancePrologBridgeProtocol",
    "VolkRegistryProtocol",
    "AuditTrailProtocol",
    "RedisStateManagerProtocol",
]