Lars
3a0ef3a734
Some checks failed
Deploy Development / deploy (push) Successful in 52s
Test Suite / pytest-backend (push) Failing after 5m3s
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
join_gate, strand_project_id, Join-Auto-Schalten, multi-active Gates, Progression-Demo und PO-Lock-Spec. FE: Today strang-gruppiert, Join-Warteliste in Kontrolle. Co-authored-by: Cursor <cursoragent@cursor.com>
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""Maturity progression NextAction strategy — AP2.0d."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from steering.signals import default_rules
|
|
from steering.strategies.next_action.default_strategy import DefaultNextActionStrategy
|
|
from steering.strategies.next_action.execution_ready import (
|
|
list_execution_ready_candidates,
|
|
merge_candidates,
|
|
)
|
|
from steering.strategies.next_action.product_milestone_driven import (
|
|
ProductMilestoneDrivenStrategy,
|
|
)
|
|
from steering.strategies.next_action.recurring_helpers import recurring_due_candidates
|
|
from steering.strategies.next_action.registry import (
|
|
get_next_action_strategy,
|
|
register_next_action_strategy,
|
|
)
|
|
from tenant_context import TenantContext
|
|
|
|
_default = DefaultNextActionStrategy()
|
|
_milestone = ProductMilestoneDrivenStrategy()
|
|
|
|
|
|
class MaturityProgressionStrategy:
|
|
key = "maturity_progression"
|
|
|
|
def evaluate(
|
|
self,
|
|
ctx: TenantContext,
|
|
*,
|
|
initiative_id: str | None = None,
|
|
limit: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
if limit < 1:
|
|
limit = 1
|
|
if not initiative_id:
|
|
return _default.evaluate(ctx, limit=limit)
|
|
|
|
recurring = recurring_due_candidates(ctx, initiative_id, limit=limit)
|
|
|
|
remaining = limit - len(recurring)
|
|
from services.progression_graph import list_active_work_gates
|
|
|
|
active_gates = list_active_work_gates(
|
|
tenant_id=ctx.tenant_id, initiative_id=initiative_id
|
|
)
|
|
gate_scope = active_gates[0]["id"] if len(active_gates) == 1 else None
|
|
ready: list[dict[str, Any]] = []
|
|
if remaining > 0:
|
|
ready = list_execution_ready_candidates(
|
|
ctx,
|
|
initiative_id,
|
|
limit=remaining,
|
|
scope_roadmap_item_id=gate_scope,
|
|
prefer_critical_path=True,
|
|
)
|
|
|
|
merged = merge_candidates(recurring, ready, limit=limit)
|
|
if len(merged) >= limit:
|
|
return merged[:limit]
|
|
|
|
gate_items: list[dict[str, Any]] = []
|
|
rest_limit = limit - len(merged)
|
|
if rest_limit > 0:
|
|
gate_items = _milestone.evaluate(
|
|
ctx, initiative_id=initiative_id, limit=rest_limit
|
|
)
|
|
gate_items = [
|
|
item for item in gate_items if item.get("kind") == "review_milestone"
|
|
]
|
|
|
|
merged = merge_candidates(merged, gate_items, limit=limit)
|
|
if len(merged) >= limit:
|
|
return merged[:limit]
|
|
|
|
rest = default_rules.get_next_action_candidates_for_initiative(
|
|
ctx, initiative_id=initiative_id, limit=limit - len(merged)
|
|
)
|
|
return merge_candidates(merged, rest, limit=limit)
|
|
|
|
|
|
maturity_progression_strategy = MaturityProgressionStrategy()
|
|
|
|
|
|
def register_maturity_progression_strategy() -> None:
|
|
if not get_next_action_strategy(maturity_progression_strategy.key):
|
|
register_next_action_strategy(maturity_progression_strategy)
|