Kairo-Jinkendo/backend/steering/lifecycle/slot_providers.py

"""Lifecycle slot providers — minimal registry v0.2."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Callable, Literal

from steering.lifecycle.event_transitions import EventKind

SlotKey = Literal[
    "result_intake",
    "validation",
    "adaptation",
    "action_selection",
    "planning",
]

TransitionResolver = Callable[
    [str, EventKind, dict[str, Any] | None],
    tuple[str, ...],
]

_PROVIDERS: dict[SlotKey, TransitionResolver] = {}


@dataclass(frozen=True)
class SlotProviderRegistration:
    slot_key: SlotKey
    event_kinds: frozenset[EventKind]


def register_slot_provider(
    slot_key: SlotKey,
    *,
    event_kinds: frozenset[EventKind],
    resolver: TransitionResolver,
) -> None:
    _PROVIDERS[slot_key] = resolver
    _REGISTRATIONS[slot_key] = SlotProviderRegistration(
        slot_key=slot_key, event_kinds=event_kinds
    )


_REGISTRATIONS: dict[SlotKey, SlotProviderRegistration] = {}


def list_slot_provider_registrations() -> tuple[SlotProviderRegistration, ...]:
    return tuple(_REGISTRATIONS.values())


def resolve_event_via_providers(
    *,
    primary_method_key: str,
    event_kind: EventKind,
    payload: dict[str, Any] | None = None,
) -> tuple[str, ...]:
    """First matching provider wins; fallback to default event_transitions."""
    from steering.lifecycle.event_transitions import resolve_lifecycle_chain

    payload = payload or {}
    for slot_key, registration in _REGISTRATIONS.items():
        if event_kind not in registration.event_kinds:
            continue
        resolver = _PROVIDERS.get(slot_key)
        if resolver:
            chain = resolver(primary_method_key, event_kind, payload)
            if chain:
                return chain
    return resolve_lifecycle_chain(
        primary_method_key=primary_method_key,
        event_kind=event_kind,
        payload=payload,
    )


def register_default_slot_providers() -> None:
    from steering.lifecycle.event_transitions import resolve_lifecycle_chain

    if _PROVIDERS:
        return

    def _default_resolver(
        primary_method_key: str,
        event_kind: EventKind,
        payload: dict[str, Any] | None,
    ) -> tuple[str, ...]:
        return resolve_lifecycle_chain(
            primary_method_key=primary_method_key,
            event_kind=event_kind,
            payload=payload,
        )

    register_slot_provider(
        "result_intake",
        event_kinds=frozenset({"actor_result"}),
        resolver=_default_resolver,
    )
    register_slot_provider(
        "validation",
        event_kinds=frozenset({"gate_verified"}),
        resolver=_default_resolver,
    )
    register_slot_provider(
        "adaptation",
        event_kinds=frozenset({"actor_result", "gate_verified", "gate_reopened"}),
        resolver=_default_resolver,
    )