Lars
75a6a63665
Some checks failed
Deploy Development / deploy (push) Successful in 48s
Test Suite / pytest-backend (push) Failing after 3m33s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
evaluate_steering() vereinheitlicht Horizon, Next Work und Attention; Snapshot und Signal Engine delegieren dorthin; slot_map deklarativ fuer sequential_dependency und generic_operating. Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""Lifecycle slot resolution — Kernel §4 (declarative, v0.1.1)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from steering.kernel.models import LifecycleContext, SteeringBinding
|
|
from steering.lifecycle.slot_maps import (
|
|
active_slot_keys,
|
|
is_slot_operational,
|
|
slot_map_from_tuple,
|
|
)
|
|
from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS, lifecycle_label
|
|
from steering.methods.registry import get_method
|
|
from services import steering_context as sc_service
|
|
from tenant_context import TenantContext
|
|
|
|
|
|
def _derive_slot_map_from_steps(steps: tuple[str, ...]) -> dict[str, str]:
|
|
active = set(steps)
|
|
return {
|
|
step: ("active" if step in active else "n/a") for step in STANDARD_LIFECYCLE_STEPS
|
|
}
|
|
|
|
|
|
def resolve_method_slot_map(method_key: str) -> dict[str, str]:
|
|
method = get_method(method_key)
|
|
if not method:
|
|
return _derive_slot_map_from_steps(STANDARD_LIFECYCLE_STEPS)
|
|
if method.slot_map:
|
|
return slot_map_from_tuple(method.slot_map)
|
|
return _derive_slot_map_from_steps(method.default_lifecycle_steps)
|
|
|
|
|
|
def resolve_lifecycle_context(
|
|
ctx: TenantContext,
|
|
*,
|
|
initiative_id: str,
|
|
binding: SteeringBinding,
|
|
) -> LifecycleContext:
|
|
slot_map = resolve_method_slot_map(binding.primary_method_key)
|
|
|
|
row = sc_service.get_steering_context(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
current_state = row["lifecycle_state"] if row else "intake"
|
|
if current_state not in STANDARD_LIFECYCLE_STEPS:
|
|
current_state = "intake"
|
|
|
|
active = active_slot_keys(slot_map) # type: ignore[arg-type]
|
|
operational = tuple(
|
|
step for step in STANDARD_LIFECYCLE_STEPS if is_slot_operational(slot_map.get(step, "n/a")) # type: ignore[arg-type]
|
|
)
|
|
|
|
return LifecycleContext(
|
|
current_state=current_state,
|
|
current_state_label=lifecycle_label(current_state),
|
|
slot_map=slot_map,
|
|
active_slots=active,
|
|
operational_slots=operational,
|
|
)
|
|
|
|
|
|
def lifecycle_to_dict(lifecycle: LifecycleContext) -> dict[str, Any]:
|
|
return lifecycle.to_dict()
|