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

"""Method slot maps — Kernel §4 / Spec-D D5."""

from __future__ import annotations

from typing import Literal

from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS

SlotStatus = Literal["active", "skip", "n/a", "optional"]

VALID_SLOT_STATUSES = frozenset({"active", "skip", "n/a", "optional"})

# Spec-D sequential_dependency D5
SEQUENTIAL_DEPENDENCY_SLOTS: dict[str, SlotStatus] = {
    "intake": "active",
    "method_selection": "active",
    "structure_setup": "active",
    "planning": "active",
    "action_selection": "active",
    "assignment": "active",
    "waiting": "active",
    "result_intake": "active",
    "validation": "active",
    "review": "optional",
    "adaptation": "active",
    "closure": "active",
}

# Spec-D generic_operating D5 (method_bind → method_selection, next_work → action_selection)
GENERIC_OPERATING_SLOTS: dict[str, SlotStatus] = {
    "intake": "active",
    "method_selection": "active",
    "structure_setup": "optional",
    "planning": "n/a",
    "action_selection": "active",
    "assignment": "active",
    "waiting": "optional",
    "result_intake": "active",
    "validation": "n/a",
    "review": "n/a",
    "adaptation": "optional",
    "closure": "optional",
}


# Spec-D continuous_product D5 (method_bind → method_selection)
CONTINUOUS_PRODUCT_SLOTS: dict[str, SlotStatus] = {
    "intake": "active",
    "method_selection": "active",
    "structure_setup": "active",
    "planning": "active",
    "action_selection": "active",
    "assignment": "active",
    "waiting": "active",
    "result_intake": "active",
    "validation": "optional",
    "review": "optional",
    "adaptation": "active",
    "closure": "n/a",
}


def validate_slot_map(slot_map: dict[str, SlotStatus]) -> None:
    for step, status in slot_map.items():
        if step not in STANDARD_LIFECYCLE_STEPS:
            raise ValueError(f"Unknown lifecycle slot: {step}")
        if status not in VALID_SLOT_STATUSES:
            raise ValueError(f"Invalid slot status for {step}: {status}")
    missing = set(STANDARD_LIFECYCLE_STEPS) - set(slot_map.keys())
    if missing:
        raise ValueError(f"slot_map missing lifecycle steps: {sorted(missing)}")


def slot_map_to_tuple(slot_map: dict[str, SlotStatus]) -> tuple[tuple[str, SlotStatus], ...]:
    validate_slot_map(slot_map)
    return tuple((step, slot_map[step]) for step in STANDARD_LIFECYCLE_STEPS)


def slot_map_from_tuple(items: tuple[tuple[str, SlotStatus], ...]) -> dict[str, SlotStatus]:
    if not items:
        return {}
    slot_map = dict(items)
    validate_slot_map(slot_map)
    return slot_map


def active_slot_keys(slot_map: dict[str, SlotStatus]) -> tuple[str, ...]:
    return tuple(step for step in STANDARD_LIFECYCLE_STEPS if slot_map.get(step) == "active")


def is_slot_operational(status: SlotStatus) -> bool:
    """Slots that participate in runtime evaluation (v0.1.1)."""
    return status in ("active", "optional")