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>
67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
"""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)
|