"""Lifecycle transition chains for kernel event ingress v0.2.""" from __future__ import annotations from typing import Any, Literal from steering.lifecycle.slot_maps import SlotStatus EventKind = Literal[ "actor_result", "gate_verified", "gate_reopened", "external_event", "cadence_due", "situational", ] def _slot_accepts_event(status: SlotStatus, *, event_targets_slot: bool) -> bool: if status == "active": return True if status == "optional" and event_targets_slot: return True return False def resolve_lifecycle_chain( *, primary_method_key: str, event_kind: EventKind, payload: dict[str, Any] | None = None, ) -> tuple[str, ...]: """ Ordered lifecycle slots to visit for an event (method-aware via slot_map). Returns slot keys; caller persists the last applicable state. """ from steering.kernel.lifecycle import resolve_method_slot_map payload = payload or {} slot_map = resolve_method_slot_map(primary_method_key) if event_kind == "actor_result" and payload.get("to_status") == "done": candidates = ("result_intake", "adaptation", "action_selection") event_slot = "result_intake" elif event_kind == "gate_verified": candidates = ("validation", "adaptation", "action_selection") event_slot = "validation" elif event_kind == "gate_reopened": candidates = ("adaptation", "planning", "action_selection") event_slot = "adaptation" elif event_kind == "situational" and payload.get("reason") == "blocker_resolved": candidates = ("action_selection",) event_slot = "action_selection" else: return () chain: list[str] = [] for step in candidates: status = slot_map.get(step, "n/a") if _slot_accepts_event( status, # type: ignore[arg-type] event_targets_slot=(step == event_slot or step in ("adaptation", "action_selection")), ): chain.append(step) return tuple(chain)