Lars
753f178b0c
Some checks failed
Deploy Development / deploy (push) Successful in 48s
Test Suite / pytest-backend (push) Failing after 3m23s
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 3s
Test Suite / compose-smoke (push) Has been skipped
Decision-Lock, Spec-D-Methoden und C1-Massstab-Vollspecs als Zielbild; Steering-Registry/Compat und Horizon-Tests an die Normierung anbinden. Implementierungsausreichendheit bleibt bewusst offen. Co-authored-by: Cursor <cursoragent@cursor.com>
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""Signal Engine — method-aware NextAction (AP1.1)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Literal
|
|
|
|
from services import steering_context as sc_service
|
|
from steering.methods.registry import get_method
|
|
from steering.signals import default_rules
|
|
from steering.strategies.next_action.default_strategy import default_strategy
|
|
from steering.strategies.next_action.registry import get_next_action_strategy
|
|
from tenant_context import TenantContext
|
|
|
|
SignalKind = Literal["attention", "next_action"]
|
|
|
|
|
|
def _resolve_method_key(ctx: TenantContext, initiative_id: str | None) -> str:
|
|
if not initiative_id:
|
|
return "generic_operating"
|
|
row = sc_service.get_steering_context(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
if row:
|
|
return row["method_key"]
|
|
return "generic_operating"
|
|
|
|
|
|
def _primary_composes_with_agile(method_key: str) -> bool:
|
|
agile = get_method("agile_iteration")
|
|
if not agile:
|
|
return False
|
|
return method_key in agile.composes_with
|
|
|
|
|
|
def _resolve_next_action_strategy_key(
|
|
ctx: TenantContext, initiative_id: str | None
|
|
) -> str:
|
|
method_key = _resolve_method_key(ctx, initiative_id)
|
|
method = get_method(method_key)
|
|
strategy_key = method.next_action_strategy_key if method else default_strategy.key
|
|
if not initiative_id:
|
|
return strategy_key
|
|
|
|
from services.work_cycle import get_active_work_cycle
|
|
|
|
active = get_active_work_cycle(tenant_id=ctx.tenant_id, initiative_id=initiative_id)
|
|
if active and _primary_composes_with_agile(method_key):
|
|
agile = get_next_action_strategy("agile_iteration")
|
|
if agile:
|
|
return agile.key
|
|
return strategy_key
|
|
|
|
|
|
def evaluate(
|
|
ctx: TenantContext,
|
|
kind: SignalKind = "attention",
|
|
*,
|
|
limit: int = 10,
|
|
initiative_id: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
if kind == "attention":
|
|
return default_rules.get_attention_items(ctx)
|
|
|
|
strategy_key = _resolve_next_action_strategy_key(ctx, initiative_id)
|
|
strategy = get_next_action_strategy(strategy_key) or default_strategy
|
|
return strategy.evaluate(ctx, initiative_id=initiative_id, limit=limit)
|