Kairo-Jinkendo/backend/steering/lifecycle/slot_maps.py
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
feat(steering): Kernel Spine v0.2 — Events, Slot-Provider, continuous_product
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>
2026-07-26 16:33:00 +02:00

94 lines
2.9 KiB
Python

"""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")