Lars
3bd983297b
Some checks failed
Deploy Development / deploy (push) Successful in 46s
Test Suite / pytest-backend (push) Failing after 2m2s
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
PO-Dokumente (ADP v0.2, MVP v0.3) und Minimal Complete Slice: Registry-Methoden, Initiative/Project-Spiegel, Default-Methode bei Anlage, Lagebild mit Archetyp und Guidance. Co-authored-by: Cursor <cursoragent@cursor.com>
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
"""Continuous product NextAction strategy — AP2.0a (Ist zuerst)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from db import get_connection
|
|
from psycopg2.extras import RealDictCursor
|
|
from steering.signals import default_rules
|
|
from steering.strategies.next_action.default_strategy import DefaultNextActionStrategy
|
|
from steering.strategies.next_action.registry import (
|
|
get_next_action_strategy,
|
|
register_next_action_strategy,
|
|
)
|
|
from tenant_context import TenantContext
|
|
|
|
_default = DefaultNextActionStrategy()
|
|
|
|
_OPEN_ACTION = ("open", "ready", "in_progress", "blocked", "review_required")
|
|
|
|
|
|
class ContinuousProductStrategy:
|
|
key = "continuous_product"
|
|
|
|
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)
|
|
|
|
candidates: list[dict[str, Any]] = []
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
'action' AS kind,
|
|
a.title AS title,
|
|
'Offenes Arbeitspaket priorisieren' AS summary,
|
|
a.initiative_id,
|
|
a.id AS action_id,
|
|
NULL::uuid AS backlog_item_id,
|
|
'open_action_priority' AS reason_code,
|
|
'Als Nächstes ausführen' AS recommended_action
|
|
FROM actions a
|
|
WHERE a.tenant_id = %s AND a.initiative_id = %s
|
|
AND a.status = ANY(%s)
|
|
ORDER BY
|
|
CASE a.status
|
|
WHEN 'blocked' THEN 0
|
|
WHEN 'review_required' THEN 1
|
|
WHEN 'in_progress' THEN 2
|
|
WHEN 'ready' THEN 3
|
|
ELSE 4
|
|
END,
|
|
CASE a.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1 ELSE 2 END,
|
|
a.due_at ASC NULLS LAST,
|
|
a.updated_at DESC
|
|
LIMIT %s
|
|
""",
|
|
(ctx.tenant_id, initiative_id, list(_OPEN_ACTION), limit),
|
|
)
|
|
for row in cur.fetchall():
|
|
item = dict(row)
|
|
item["initiative_id"] = str(item["initiative_id"])
|
|
item["action_id"] = str(item["action_id"])
|
|
candidates.append(item)
|
|
finally:
|
|
conn.close()
|
|
|
|
remaining = limit - len(candidates)
|
|
if remaining > 0:
|
|
from steering.strategies.next_action.product_milestone_driven import (
|
|
ProductMilestoneDrivenStrategy,
|
|
)
|
|
|
|
gate_items = ProductMilestoneDrivenStrategy().evaluate(
|
|
ctx, initiative_id=initiative_id, limit=remaining
|
|
)
|
|
for item in gate_items:
|
|
if item.get("kind") == "review_milestone":
|
|
candidates.append(item)
|
|
remaining -= 1
|
|
if remaining <= 0:
|
|
break
|
|
|
|
if len(candidates) < limit:
|
|
rest = default_rules.get_next_action_candidates_for_initiative(
|
|
ctx, initiative_id=initiative_id, limit=limit - len(candidates)
|
|
)
|
|
seen = {c.get("action_id") for c in candidates if c.get("action_id")}
|
|
for item in rest:
|
|
if item.get("action_id") and item["action_id"] in seen:
|
|
continue
|
|
candidates.append(item)
|
|
if len(candidates) >= limit:
|
|
break
|
|
|
|
return candidates[:limit]
|
|
|
|
|
|
continuous_product_strategy = ContinuousProductStrategy()
|
|
|
|
|
|
def register_continuous_product_strategy() -> None:
|
|
if not get_next_action_strategy(continuous_product_strategy.key):
|
|
register_next_action_strategy(continuous_product_strategy)
|