Lars
fe1edb84f8
All checks were successful
Deploy Development / deploy (push) Successful in 50s
Test Suite / pytest-backend (push) Successful in 3m49s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 19s
Test Suite / playwright-smoke (push) Successful in 12s
apply_steering_event an Action done, Gate verify/reopen und Blocker resolve; lifecycle-Ketten methodenaware via slot_map; continuous_product slot_map; pytest v0.2. Co-authored-by: Cursor <cursoragent@cursor.com>
108 lines
2.8 KiB
Python
108 lines
2.8 KiB
Python
"""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,
|
|
)
|