AP2.0d: Next-Action-Strategien nutzen Execution-Graph ready_actions.

Archetyp-Strategien priorisieren ausführungsbereite APs mit Begründung execution_ready/critical_path; continuous_product und program_delivery integriert.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/steering/methods/registrations/ap20_method_stubs.py

@@ -10,25 +10,25 @@ def register() -> None:
         key="maturity_progression",
         label="Reifegrad-Entwicklung",
         description="Stufen, Routinen, historische Entwicklung",
-        next_action_strategy_key="default",
+        next_action_strategy_key="maturity_progression",
     )
     register_stub_method(
         key="sequential_dependency",
         label="Sequenzielle Abhängigkeit",
         description="Graph-Pfade, kritischer Pfad (read model)",
-        next_action_strategy_key="default",
+        next_action_strategy_key="sequential_dependency",
     )
     register_stub_method(
         key="recurring_control",
         label="Rhythmus-Steuerung",
         description="Dauerprogramm, fällig/überfällig, Abweichungen",
-        next_action_strategy_key="default",
+        next_action_strategy_key="recurring_control",
     )
     register_stub_method(
         key="queue_pull",
         label="Inbox / Queue",
         description="Pull oder Empfehlung aus Queue",
-        next_action_strategy_key="default",
+        next_action_strategy_key="queue_pull",
     )
     register_stub_method(
         key="agile_iteration",

backend/steering/strategies/next_action/__init__.py

@@ -7,6 +7,9 @@ from steering.strategies.next_action.continuous_product import (
     register_continuous_product_strategy,
 )
 from steering.strategies.next_action.default_strategy import default_strategy
+from steering.strategies.next_action.maturity_progression import (
+    register_maturity_progression_strategy,
+)
 from steering.strategies.next_action.product_milestone_driven import (
     product_milestone_strategy,
 )
@@ -14,10 +17,17 @@ from steering.strategies.next_action.program_delivery import (
     program_delivery_strategy,
     register_program_delivery_strategy,
 )
+from steering.strategies.next_action.queue_pull import register_queue_pull_strategy
+from steering.strategies.next_action.recurring_control import (
+    register_recurring_control_strategy,
+)
 from steering.strategies.next_action.registry import (
     get_next_action_strategy,
     register_next_action_strategy,
 )
+from steering.strategies.next_action.sequential_dependency import (
+    register_sequential_dependency_strategy,
+)
 
 
 def register_builtin_strategies() -> None:
@@ -26,3 +36,7 @@ def register_builtin_strategies() -> None:
             register_next_action_strategy(strategy)
     register_program_delivery_strategy()
     register_continuous_product_strategy()
+    register_sequential_dependency_strategy()
+    register_maturity_progression_strategy()
+    register_recurring_control_strategy()
+    register_queue_pull_strategy()

backend/steering/strategies/next_action/continuous_product.py

@@ -1,4 +1,4 @@
-"""Continuous product NextAction strategy — AP2.0a (Ist zuerst)."""
+"""Continuous product NextAction strategy — AP2.0a + AP2.0d."""
 
 from __future__ import annotations
 
@@ -8,6 +8,11 @@ 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.execution_ready import (
+    blocked_execution_action_ids,
+    list_execution_ready_candidates,
+    merge_candidates,
+)
 from steering.strategies.next_action.registry import (
     get_next_action_strategy,
     register_next_action_strategy,
@@ -34,12 +39,26 @@ class ContinuousProductStrategy:
         if not initiative_id:
             return _default.evaluate(ctx, limit=limit)
 
-        candidates: list[dict[str, Any]] = []
+        blocked_ids = blocked_execution_action_ids(ctx, initiative_id)
+        ready = list_execution_ready_candidates(
+            ctx, initiative_id, limit=limit, prefer_critical_path=True
+        )
+
+        remaining = limit - len(ready)
+        operational: list[dict[str, Any]] = []
+        if remaining > 0:
+            blocked_list = list(blocked_ids)
+            blocked_clause = ""
+            blocked_params: list[Any] = []
+            if blocked_list:
+                blocked_clause = "AND NOT (a.id::text = ANY(%s))"
+                blocked_params = [blocked_list]
+
             conn = get_connection()
             try:
                 with conn.cursor(cursor_factory=RealDictCursor) as cur:
                     cur.execute(
-                    """
+                        f"""
                         SELECT
                             'action' AS kind,
                             a.title AS title,
@@ -52,30 +71,46 @@ class ContinuousProductStrategy:
                         FROM actions a
                         WHERE a.tenant_id = %s AND a.initiative_id = %s
                           AND a.status = ANY(%s)
+                          {blocked_clause}
                         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
+                                WHEN 'review_required' THEN 0
+                                WHEN 'in_progress' THEN 1
+                                WHEN 'ready' THEN 2
+                                ELSE 3
+                            END,
+                            CASE a.priority
+                                WHEN 'critical' THEN 0
+                                WHEN 'high' THEN 1
+                                ELSE 2
                             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),
+                        (
+                            ctx.tenant_id,
+                            initiative_id,
+                            list(_OPEN_ACTION),
+                            *blocked_params,
+                            remaining,
+                        ),
                     )
                     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)
+                        if item["action_id"] in blocked_ids:
+                            continue
+                        operational.append(item)
             finally:
                 conn.close()
 
-        remaining = limit - len(candidates)
+        merged = merge_candidates(
+            ready, operational, limit=limit, exclude_action_ids=blocked_ids
+        )
+
+        remaining = limit - len(merged)
         if remaining > 0:
             from steering.strategies.next_action.product_milestone_driven import (
                 ProductMilestoneDrivenStrategy,
@@ -84,26 +119,22 @@ class ContinuousProductStrategy:
             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)
+            gate_items = [
+                item for item in gate_items if item.get("kind") == "review_milestone"
+            ]
+            merged = merge_candidates(
+                merged, gate_items, limit=limit, exclude_action_ids=blocked_ids
             )
-            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]
+        if len(merged) < limit:
+            rest = default_rules.get_next_action_candidates_for_initiative(
+                ctx, initiative_id=initiative_id, limit=limit - len(merged)
+            )
+            merged = merge_candidates(
+                merged, rest, limit=limit, exclude_action_ids=blocked_ids
+            )
+
+        return merged[:limit]
 
 
 continuous_product_strategy = ContinuousProductStrategy()

backend/steering/strategies/next_action/execution_ready.py

@@ -0,0 +1,195 @@
+"""Execution-ready ranking for NextAction strategies — AP2.0d."""
+
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+from steering.graph.execution_engine import load_initiative_execution_graph_state
+from tenant_context import TenantContext
+
+_OPEN_ACTION = frozenset({"open", "ready", "in_progress", "blocked", "review_required"})
+
+
+def rank_ready_action_ids(
+    graph_state: dict[str, Any],
+    *,
+    prefer_critical_path: bool = True,
+) -> list[str]:
+    """Order ready_actions: critical path first, then sort_order."""
+    ready = [str(action_id) for action_id in graph_state.get("ready_actions", [])]
+    if not ready:
+        return []
+
+    items = graph_state.get("items") or {}
+    ready_set = set(ready)
+    ordered: list[str] = []
+
+    if prefer_critical_path:
+        for action_id in graph_state.get("critical_path") or []:
+            aid = str(action_id)
+            if aid in ready_set and aid not in ordered:
+                ordered.append(aid)
+
+    remaining = [aid for aid in ready if aid not in ordered]
+    remaining.sort(
+        key=lambda action_id: (
+            items.get(action_id, {}).get("sort_order", 0),
+            action_id,
+        )
+    )
+    ordered.extend(remaining)
+    return ordered
+
+
+def action_to_next_candidate(
+    action: dict[str, Any],
+    *,
+    reason_code: str = "execution_ready",
+    summary: str = "Ausführungsbereit — keine blockierenden Vorgänger",
+    recommended_action: str = "Als Nächstes ausführen",
+    on_critical_path: bool = False,
+) -> dict[str, Any]:
+    """Convert action row to NextAction candidate."""
+    if on_critical_path and reason_code == "execution_ready":
+        reason_code = "execution_critical_path"
+        summary = "Kritischer Pfad — als Nächstes ausführen"
+
+    return {
+        "kind": "action",
+        "title": action.get("title") or "Arbeitspaket",
+        "summary": summary,
+        "initiative_id": str(action["initiative_id"]),
+        "action_id": str(action["id"]),
+        "backlog_item_id": None,
+        "reason_code": reason_code,
+        "recommended_action": recommended_action,
+    }
+
+
+def build_execution_ready_candidates(
+    *,
+    actions: list[dict[str, Any]],
+    graph_state: dict[str, Any],
+    limit: int,
+    prefer_critical_path: bool = True,
+    reason_code: str = "execution_ready",
+) -> list[dict[str, Any]]:
+    """Build ranked NextAction candidates from a preloaded graph state."""
+    if limit < 1:
+        return []
+
+    action_by_id = {str(action["id"]): action for action in actions}
+    critical_path = {str(action_id) for action_id in graph_state.get("critical_path") or []}
+    blocked_ids = {str(action_id) for action_id in graph_state.get("blocked_actions") or []}
+
+    candidates: list[dict[str, Any]] = []
+    for action_id in rank_ready_action_ids(
+        graph_state, prefer_critical_path=prefer_critical_path
+    ):
+        action = action_by_id.get(action_id)
+        if not action:
+            continue
+        if action.get("status") not in _OPEN_ACTION:
+            continue
+        if action_id in blocked_ids:
+            continue
+        candidates.append(
+            action_to_next_candidate(
+                action,
+                reason_code=reason_code,
+                on_critical_path=action_id in critical_path,
+            )
+        )
+        if len(candidates) >= limit:
+            break
+    return candidates
+
+
+def list_execution_ready_candidates(
+    ctx: TenantContext,
+    initiative_id: str,
+    *,
+    limit: int,
+    scope_roadmap_item_id: str | None = None,
+    prefer_critical_path: bool = True,
+    reason_code: str = "execution_ready",
+) -> list[dict[str, Any]]:
+    """Load execution graph and return ranked ready-action candidates."""
+    from services import actions as action_service
+
+    actions = action_service.list_actions_for_initiative(
+        tenant_id=ctx.tenant_id, initiative_id=initiative_id
+    )
+    graph_state = load_initiative_execution_graph_state(
+        tenant_id=ctx.tenant_id,
+        initiative_id=initiative_id,
+        scope_roadmap_item_id=scope_roadmap_item_id,
+    )
+    return build_execution_ready_candidates(
+        actions=actions,
+        graph_state=graph_state,
+        limit=limit,
+        prefer_critical_path=prefer_critical_path,
+        reason_code=reason_code,
+    )
+
+
+def merge_candidates(
+    *sources: Iterable[dict[str, Any]],
+    limit: int,
+    exclude_action_ids: set[str] | None = None,
+) -> list[dict[str, Any]]:
+    """Merge candidate lists without duplicate action_ids."""
+    excluded = exclude_action_ids or set()
+    merged: list[dict[str, Any]] = []
+    seen_actions: set[str] = set()
+
+    for source in sources:
+        for item in source:
+            action_id = item.get("action_id")
+            if action_id:
+                if action_id in excluded or action_id in seen_actions:
+                    continue
+                seen_actions.add(action_id)
+            merged.append(item)
+            if len(merged) >= limit:
+                return merged
+    return merged
+
+
+def blocked_execution_action_ids(
+    ctx: TenantContext, initiative_id: str, *, scope_roadmap_item_id: str | None = None
+) -> set[str]:
+    """Action IDs blocked by execution graph (not necessarily status=blocked)."""
+    graph_state = load_initiative_execution_graph_state(
+        tenant_id=ctx.tenant_id,
+        initiative_id=initiative_id,
+        scope_roadmap_item_id=scope_roadmap_item_id,
+    )
+    return {str(action_id) for action_id in graph_state.get("blocked_actions") or []}
+
+
+def first_active_gate_id(ctx: TenantContext, initiative_id: str) -> str | None:
+    """First active roadmap item for initiative scope (Gate-Horizont)."""
+    from db import get_connection
+    from psycopg2.extras import RealDictCursor
+
+    conn = get_connection()
+    try:
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT ri.id
+                FROM roadmap_items ri
+                JOIN roadmaps r ON r.id = ri.roadmap_id AND r.tenant_id = ri.tenant_id
+                WHERE ri.tenant_id = %s AND r.initiative_id = %s
+                  AND ri.status = 'active'
+                ORDER BY ri.target_date ASC NULLS LAST, ri.updated_at DESC
+                LIMIT 1
+                """,
+                (ctx.tenant_id, initiative_id),
+            )
+            row = cur.fetchone()
+            return str(row["id"]) if row else None
+    finally:
+        conn.close()

backend/steering/strategies/next_action/maturity_progression.py

@@ -0,0 +1,76 @@
+"""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 (
+    first_active_gate_id,
+    list_execution_ready_candidates,
+    merge_candidates,
+)
+from steering.strategies.next_action.product_milestone_driven import (
+    ProductMilestoneDrivenStrategy,
+)
+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)
+
+        gate_scope = first_active_gate_id(ctx, initiative_id)
+        ready = list_execution_ready_candidates(
+            ctx,
+            initiative_id,
+            limit=limit,
+            scope_roadmap_item_id=gate_scope,
+            prefer_critical_path=True,
+        )
+
+        remaining = limit - len(ready)
+        gate_items: list[dict[str, Any]] = []
+        if remaining > 0:
+            gate_items = _milestone.evaluate(
+                ctx, initiative_id=initiative_id, limit=remaining
+            )
+            gate_items = [
+                item for item in gate_items if item.get("kind") == "review_milestone"
+            ]
+
+        merged = merge_candidates(ready, 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)

backend/steering/strategies/next_action/program_delivery.py

@@ -1,7 +1,16 @@
-"""Program delivery NextAction strategy — AP2.0a (Gate-Horizont)."""
+"""Program delivery NextAction strategy — AP2.0a + AP2.0d (Gate-Horizont)."""
 
 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 (
+    first_active_gate_id,
+    list_execution_ready_candidates,
+    merge_candidates,
+)
 from steering.strategies.next_action.product_milestone_driven import (
     ProductMilestoneDrivenStrategy,
 )
@@ -9,15 +18,54 @@ from steering.strategies.next_action.registry import (
     get_next_action_strategy,
     register_next_action_strategy,
 )
+from tenant_context import TenantContext
 
-_strategy = ProductMilestoneDrivenStrategy()
+_default = DefaultNextActionStrategy()
+_milestone = ProductMilestoneDrivenStrategy()
 
 
 class ProgramDeliveryStrategy:
     key = "program_delivery"
 
-    def evaluate(self, ctx, *, initiative_id=None, limit=10):
-        return _strategy.evaluate(ctx, initiative_id=initiative_id, limit=limit)
+    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)
+
+        gate_scope = first_active_gate_id(ctx, initiative_id)
+        ready = list_execution_ready_candidates(
+            ctx,
+            initiative_id,
+            limit=limit,
+            scope_roadmap_item_id=gate_scope,
+            prefer_critical_path=True,
+        )
+
+        remaining = limit - len(ready)
+        gate_items: list[dict[str, Any]] = []
+        if remaining > 0:
+            gate_items = _milestone.evaluate(
+                ctx, initiative_id=initiative_id, limit=remaining
+            )
+            gate_items = [
+                item for item in gate_items if item.get("kind") == "review_milestone"
+            ]
+
+        merged = merge_candidates(ready, 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)
 
 
 program_delivery_strategy = ProgramDeliveryStrategy()

backend/steering/strategies/next_action/queue_pull.py

@@ -0,0 +1,99 @@
+"""Queue pull NextAction strategy — AP2.0d."""
+
+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.execution_ready import (
+    list_execution_ready_candidates,
+    merge_candidates,
+)
+from steering.strategies.next_action.registry import (
+    get_next_action_strategy,
+    register_next_action_strategy,
+)
+from tenant_context import TenantContext
+
+_default = DefaultNextActionStrategy()
+
+
+def _queue_backlog_candidates(
+    ctx: TenantContext, initiative_id: str, *, limit: int
+) -> list[dict[str, Any]]:
+    conn = get_connection()
+    try:
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT
+                    'convert_backlog' AS kind,
+                    bi.title AS title,
+                    'Freigegebenes Backlog-Item aus Queue ziehen' AS summary,
+                    bi.initiative_id,
+                    NULL::uuid AS action_id,
+                    bi.id AS backlog_item_id,
+                    'backlog_accepted_not_converted' AS reason_code,
+                    'In Maßnahme umwandeln' AS recommended_action
+                FROM backlog_items bi
+                WHERE bi.tenant_id = %s AND bi.initiative_id = %s
+                  AND bi.status = 'accepted'
+                  AND bi.converted_action_id IS NULL
+                ORDER BY bi.updated_at ASC
+                LIMIT %s
+                """,
+                (ctx.tenant_id, initiative_id, limit),
+            )
+            items: list[dict[str, Any]] = []
+            for row in cur.fetchall():
+                item = dict(row)
+                item["initiative_id"] = str(item["initiative_id"])
+                item["backlog_item_id"] = str(item["backlog_item_id"])
+                items.append(item)
+            return items
+    finally:
+        conn.close()
+
+
+class QueuePullStrategy:
+    key = "queue_pull"
+
+    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)
+
+        queue = _queue_backlog_candidates(ctx, initiative_id, limit=limit)
+        remaining = limit - len(queue)
+        ready: list[dict[str, Any]] = []
+        if remaining > 0:
+            ready = list_execution_ready_candidates(
+                ctx, initiative_id, limit=remaining, prefer_critical_path=False
+            )
+
+        merged = merge_candidates(queue, ready, 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)
+
+
+queue_pull_strategy = QueuePullStrategy()
+
+
+def register_queue_pull_strategy() -> None:
+    if not get_next_action_strategy(queue_pull_strategy.key):
+        register_next_action_strategy(queue_pull_strategy)

backend/steering/strategies/next_action/recurring_control.py

@@ -0,0 +1,99 @@
+"""Recurring control NextAction strategy — AP2.0d."""
+
+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.execution_ready import (
+    list_execution_ready_candidates,
+    merge_candidates,
+)
+from steering.strategies.next_action.registry import (
+    get_next_action_strategy,
+    register_next_action_strategy,
+)
+from tenant_context import TenantContext
+
+_default = DefaultNextActionStrategy()
+
+
+def _recurring_due_candidates(
+    ctx: TenantContext, initiative_id: str, *, limit: int
+) -> list[dict[str, Any]]:
+    conn = get_connection()
+    try:
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT
+                    'recurring_due' AS kind,
+                    re.title AS title,
+                    'Wiederkehrendes Element fällig' AS summary,
+                    re.initiative_id,
+                    NULL::uuid AS action_id,
+                    NULL::uuid AS backlog_item_id,
+                    'recurring_due' AS reason_code,
+                    'Rhythmus-Element bearbeiten' AS recommended_action
+                FROM recurring_elements re
+                WHERE re.tenant_id = %s AND re.initiative_id = %s
+                  AND re.status = 'active'
+                  AND re.next_due_at IS NOT NULL
+                  AND re.next_due_at <= NOW()
+                ORDER BY re.next_due_at ASC
+                LIMIT %s
+                """,
+                (ctx.tenant_id, initiative_id, limit),
+            )
+            items: list[dict[str, Any]] = []
+            for row in cur.fetchall():
+                item = dict(row)
+                item["initiative_id"] = str(item["initiative_id"])
+                items.append(item)
+            return items
+    finally:
+        conn.close()
+
+
+class RecurringControlStrategy:
+    key = "recurring_control"
+
+    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)
+        ready: list[dict[str, Any]] = []
+        if remaining > 0:
+            ready = list_execution_ready_candidates(
+                ctx, initiative_id, limit=remaining, prefer_critical_path=False
+            )
+
+        merged = merge_candidates(recurring, ready, 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)
+
+
+recurring_control_strategy = RecurringControlStrategy()
+
+
+def register_recurring_control_strategy() -> None:
+    if not get_next_action_strategy(recurring_control_strategy.key):
+        register_next_action_strategy(recurring_control_strategy)

backend/steering/strategies/next_action/sequential_dependency.py

@@ -0,0 +1,67 @@
+"""Sequential dependency 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.registry import (
+    get_next_action_strategy,
+    register_next_action_strategy,
+)
+from tenant_context import TenantContext
+
+_default = DefaultNextActionStrategy()
+
+
+class SequentialDependencyStrategy:
+    key = "sequential_dependency"
+
+    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)
+
+        ready = list_execution_ready_candidates(
+            ctx,
+            initiative_id,
+            limit=limit,
+            prefer_critical_path=True,
+        )
+        if len(ready) >= limit:
+            return ready[:limit]
+
+        from steering.strategies.next_action.execution_ready import (
+            blocked_execution_action_ids,
+        )
+
+        rest = default_rules.get_next_action_candidates_for_initiative(
+            ctx, initiative_id=initiative_id, limit=limit - len(ready)
+        )
+        blocked_ids = blocked_execution_action_ids(ctx, initiative_id)
+        filtered = [
+            item
+            for item in rest
+            if not item.get("action_id") or item["action_id"] not in blocked_ids
+        ]
+        return merge_candidates(ready, filtered, limit=limit)
+
+
+sequential_dependency_strategy = SequentialDependencyStrategy()
+
+
+def register_sequential_dependency_strategy() -> None:
+    if not get_next_action_strategy(sequential_dependency_strategy.key):
+        register_next_action_strategy(sequential_dependency_strategy)

backend/tests/test_ap20d_next_action_strategies.py

@@ -0,0 +1,136 @@
+"""Unit tests for AP2.0d NextAction execution-ready strategies."""
+
+from __future__ import annotations
+
+from steering.graph.execution_engine import compute_execution_graph_state
+from steering.methods.registry import list_methods
+from steering.strategies.next_action import register_builtin_strategies
+from steering.strategies.next_action.execution_ready import (
+    build_execution_ready_candidates,
+    merge_candidates,
+    rank_ready_action_ids,
+)
+from steering.strategies.next_action.registry import get_next_action_strategy
+
+
+def _action(action_id: str, status: str = "open", **kwargs):
+    return {
+        "id": action_id,
+        "initiative_id": kwargs.get("initiative_id", "init-1"),
+        "status": status,
+        "sort_order": kwargs.get("sort_order", 0),
+        "title": kwargs.get("title", action_id),
+        "action_kind": kwargs.get("action_kind", "delivery"),
+        "roadmap_item_id": kwargs.get("roadmap_item_id"),
+    }
+
+
+def test_rank_ready_prefers_critical_path():
+    actions = [
+        _action("a", "done", sort_order=0),
+        _action("b", "open", sort_order=1),
+        _action("c", "open", sort_order=2),
+    ]
+    deps = [
+        {
+            "predecessor_action_id": "a",
+            "successor_action_id": "b",
+            "dependency_kind": "requires",
+        },
+        {
+            "predecessor_action_id": "b",
+            "successor_action_id": "c",
+            "dependency_kind": "requires",
+        },
+    ]
+    state = compute_execution_graph_state(actions=actions, dependencies=deps)
+    ranked = rank_ready_action_ids(state, prefer_critical_path=True)
+    assert ranked == ["b"]
+
+
+def test_build_execution_ready_skips_blocked():
+    actions = [
+        _action("first", "open", sort_order=0),
+        _action("second", "open", sort_order=1),
+    ]
+    state = compute_execution_graph_state(actions=actions, dependencies=[])
+    candidates = build_execution_ready_candidates(
+        actions=actions, graph_state=state, limit=5
+    )
+    assert len(candidates) == 1
+    assert candidates[0]["action_id"] == "first"
+    assert candidates[0]["reason_code"] in ("execution_ready", "execution_critical_path")
+
+
+def test_critical_path_reason_code():
+    actions = [
+        _action("a", "done", sort_order=0),
+        _action("b", "open", sort_order=1),
+        _action("c", "open", sort_order=2),
+    ]
+    deps = [
+        {
+            "predecessor_action_id": "a",
+            "successor_action_id": "b",
+            "dependency_kind": "requires",
+        },
+        {
+            "predecessor_action_id": "b",
+            "successor_action_id": "c",
+            "dependency_kind": "requires",
+        },
+    ]
+    state = compute_execution_graph_state(actions=actions, dependencies=deps)
+    candidates = build_execution_ready_candidates(
+        actions=actions, graph_state=state, limit=5, prefer_critical_path=True
+    )
+    assert candidates[0]["reason_code"] == "execution_critical_path"
+
+
+def test_merge_candidates_deduplicates_actions():
+    first = [
+        {
+            "kind": "action",
+            "action_id": "a1",
+            "title": "A",
+        }
+    ]
+    second = [
+        {
+            "kind": "action",
+            "action_id": "a1",
+            "title": "A duplicate",
+        },
+        {
+            "kind": "review_milestone",
+            "title": "Gate",
+        },
+    ]
+    merged = merge_candidates(first, second, limit=5)
+    assert len(merged) == 2
+    assert merged[0]["action_id"] == "a1"
+    assert merged[1]["kind"] == "review_milestone"
+
+
+def test_ap20d_strategies_registered():
+    register_builtin_strategies()
+    for key in (
+        "sequential_dependency",
+        "maturity_progression",
+        "recurring_control",
+        "queue_pull",
+        "continuous_product",
+        "program_delivery",
+    ):
+        assert get_next_action_strategy(key) is not None
+
+
+def test_ap20d_methods_wire_strategies():
+    register_builtin_strategies()
+    methods = {m.key: m for m in list_methods()}
+    assert methods["sequential_dependency"].next_action_strategy_key == "sequential_dependency"
+    assert methods["maturity_progression"].next_action_strategy_key == "maturity_progression"
+    assert methods["recurring_control"].next_action_strategy_key == "recurring_control"
+    assert methods["queue_pull"].next_action_strategy_key == "queue_pull"
+    assert methods["continuous_product"].next_action_strategy_key == "continuous_product"
+    assert methods["program_delivery"].next_action_strategy_key == "program_delivery"

docs/product/Kairo_Corrected_MVP_Roadmap_v0.2.md

@@ -182,7 +182,7 @@ Siehe **`Kairo_Status_Review_and_Next_Steps_v0.1.md` §5** für vollständige Ro
 | D1 | Dogfooding R1 — Kairo-Jinkendo in Kairo | **→ nächstes** | B2b-Validation |
 | 1 | AP1.9c Cockpit-Signale | **◐ Code** | Portfolio-Kacheln via Attention |
 | 2 | AP1.16a–b Execution-Graph (Schema + Engine) | **◐ Code** | **vor** AP2.0d; Remote-Verifikation nach Deploy |
-| 3 | AP2.0d Next-Action-Strategien | offen | MVP Stufe A; nutzt `ready_actions` |
+| 3 | AP2.0d Next-Action-Strategien | **◐ Code** | MVP Stufe A; nutzt `ready_actions`; Remote-Verifikation nach Deploy |
 | 4 | AP1.16c–d Plan-Outline-Kanten + Planning Debt | offen | nach 16a–b |
 | 5 | Dogfooding R2 — Ist + Gitea-Evidence | offen | Fortschritt sichtbar |
 | 6 | AP0.10d / AP2.1 Validation B2b | offen | MVP-Urteil |

docs/product/Kairo_Implementation_Truth_Table_v0.1.md

@@ -79,7 +79,7 @@ Verhindert, dass Zielbild-Dokumente als Ist-Stand gelesen werden.
 | Signal Engine | ◐ | Regeln begrenzt; Cockpit-Kacheln offen AP1.9c |
 | Method Registry | ◐ | AP2.0a: Stubs inkl. `continuous_product`, `program_delivery`, …; 2 Strategien aktiv |
 | Archetyp-/Methoden-Katalog PO | ✓ | ADP v0.2 + MVP v0.3 |
-| NextActionCandidate | ◐ | API + Widget; `reason_code` teilweise AP2.0c; Strategien AP2.0d offen |
+| NextActionCandidate | ◐ | API + Widget; `reason_code` AP2.0c/d; Strategien nutzen `ready_actions` (AP2.0d) |
 | Portfolio-Priorität (Initiativen) | ◐ | AP1.8a: `portfolio_rank`, Reorder, Next-Action-Sort |
 | Situativer Steuerungskontext (Next Action) | ✗ | 📄 Vision §7.6; AP1.8b deferred |
 | AttentionItem | ◐ | |
@@ -151,7 +151,7 @@ Verhindert, dass Zielbild-Dokumente als Ist-Stand gelesen werden.
 | Default-Methode bei Anlage | ◐ | AP2.0b |
 | Next-Action program_delivery | ◐ | Strategie registriert |
 | Next-Action continuous_product | ◐ | Strategie registriert |
-| maturity / sequential / queue / recurring Strategien | ○ | AP2.0d |
+| maturity / sequential / queue / recurring Strategien | ◐ | AP2.0d Code; Remote-Verifikation nach Deploy |
 | Referenz-Ausprägung product.kairo_dev | ◐ | Code-Seed; Dogfooding manuell |
 | MVP-Abnahfe Stufe A validiert | ✗ | AP0.10d / AP2.1 offen |
 | Operational Actor API | ✗ | AP1.7 |
@@ -205,7 +205,7 @@ Details: `Kairo_Status_Review_and_Next_Steps_v0.1.md` §2.1
 | AP1.16a–b | Action deps, Execution Engine ✗→◐ (Deploy) |
 | AP1.16c–d | Outline-Kanten, Planning Debt ✗→◐ |
 | AP1.9c | Cockpit-Signale ◐→✓ |
-| AP2.0d | Strategien ○→◐ |
+| AP2.0d | Strategien ◐ Code |
 | Dogfooding R1 | Referenz-Vorhaben in Alltag ◐ |
 | AP2.0f | work_cycle ○→◐ |
 | AP1.7 | Operational API ✗→◐ |

docs/sprints/Sprint1_AP1_16_Execution_Plan_v0.1.md

@@ -84,7 +84,7 @@ Kanten-Richtung in DB: `predecessor_action_id` → `successor_action_id` (Vorgä
 | 14 | Attention: `planning_debt`, `execution_waiting` |
 | 15 | Cockpit Portfolio-Kacheln: Attention-Chips (AP1.9c) |
 | 16 | Method Profile `planning_levels` | deferred |
-| 17 | AP2.0d: Next-Action nutzt `ready_actions` | offen |
+| 17 | AP2.0d: Next-Action nutzt `ready_actions` | ◐ Code |
 
 ---
 

frontend/src/constants/operating.js

@@ -61,11 +61,13 @@ export const SIGNAL_LABELS = {
 }
 
 export const NEXT_ACTION_KIND_LABELS = {
+  action: 'Arbeitspaket',
   resolve_blocker: 'Blocker klären',
   assign_action: 'Maßnahme zuweisen',
   convert_backlog: 'Backlog umwandeln',
   create_action: 'Maßnahme anlegen',
   review_milestone: 'Meilenstein prüfen',
+  recurring_due: 'Rhythmus fällig',
 }
 
 export const ATTENTION_KIND_LABELS = {