AP2.3e: Method Plugin Contract mit data_slices und Archetyp-Kompatibilitaetspruefung.

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

backend/services/steering_context.py

@@ -217,12 +217,24 @@ def update_method_key(
     method_key: str,
     user_id: Optional[str] = None,
 ) -> dict[str, Any]:
-    from steering.methods.registry import get_method
+    from entity_archetypes.registry import resolve_default_method_key
+    from method_profiles.registry import get_method_profile
+    from steering.methods.registry import get_method, method_compatible_with_archetype
+    from services.initiatives import get_initiative
 
     method = get_method(method_key)
     if not method:
         raise ValueError(f"Unbekannte Methode: {method_key}")
 
+    initiative = get_initiative(tenant_id=tenant_id, initiative_id=initiative_id)
+    if not initiative:
+        raise ValueError("Vorhaben nicht gefunden")
+    if not method_compatible_with_archetype(method, initiative["archetype_key"]):
+        raise ValueError(
+            f"Methode {method_key} ist nicht kompatibel mit Archetyp "
+            f"{initiative['archetype_key']}"
+        )
+
     existing = get_steering_context(tenant_id=tenant_id, initiative_id=initiative_id)
     if not existing:
         raise ValueError("SteeringContext nicht gefunden")

backend/steering/methods/registrations/_helpers.py

@@ -1,7 +1,9 @@
-"""Shared registration helper for AP2.0 method stubs."""
+"""Shared registration helper for AP2.0 method stubs — AP2.3e data_slices."""
 
 from __future__ import annotations
 
+from typing import Literal
+
 from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS
 from steering.methods.registry import MethodDefinition, get_method, register_method
 
@@ -31,6 +33,47 @@ _LIGHT_STEPS = (
     "closure",
 )
 
+SLICES_OM_STANDARD = frozenset(
+    {
+        "backlog",
+        "actions",
+        "roadmap",
+        "projects",
+        "blockers",
+        "evidence",
+        "decisions",
+        "reviews",
+        "steering_methods",
+    }
+)
+
+SLICES_PRODUCT = SLICES_OM_STANDARD | frozenset({"work_cycles"})
+
+SLICES_QUEUE = frozenset({"backlog", "actions", "blockers", "steering_methods"})
+
+SLICES_MATURITY = frozenset(
+    {
+        "actions",
+        "roadmap",
+        "blockers",
+        "evidence",
+        "decisions",
+        "reviews",
+        "recurring",
+        "steering_methods",
+    }
+)
+
+SLICES_RECURRING = frozenset(
+    {"actions", "roadmap", "recurring", "blockers", "steering_methods"}
+)
+
+SLICES_PROGRAM = SLICES_OM_STANDARD | frozenset({"work_cycles"})
+
+SLICES_GENERIC = SLICES_OM_STANDARD
+
+SLICES_AGILE = SLICES_PRODUCT
+
 
 def register_stub_method(
     *,
@@ -39,6 +82,8 @@ def register_stub_method(
     description: str,
     next_action_strategy_key: str = "default",
     lifecycle_steps: tuple[str, ...] | None = None,
+    data_slices: frozenset[str] | None = None,
+    compatible_archetype_keys: frozenset[str] | Literal["*"] = "*",
 ) -> None:
     if get_method(key):
         return
@@ -50,6 +95,8 @@ def register_stub_method(
             description=description,
             default_lifecycle_steps=lifecycle_steps or STANDARD_LIFECYCLE_STEPS,
             next_action_strategy_key=next_action_strategy_key,
+            data_slices=data_slices or SLICES_GENERIC,
+            compatible_archetype_keys=compatible_archetype_keys,
         )
     )
 
@@ -60,6 +107,8 @@ def register_product_like_method(
     label: str,
     description: str,
     next_action_strategy_key: str,
+    data_slices: frozenset[str] | None = None,
+    compatible_archetype_keys: frozenset[str] | Literal["*"] | None = None,
 ) -> None:
     register_stub_method(
         key=key,
@@ -67,4 +116,8 @@ def register_product_like_method(
         description=description,
         next_action_strategy_key=next_action_strategy_key,
         lifecycle_steps=_PRODUCT_LIKE_STEPS,
+        data_slices=data_slices or SLICES_PRODUCT,
+        compatible_archetype_keys=compatible_archetype_keys
+        if compatible_archetype_keys is not None
+        else frozenset({"initiative.product"}),
     )

backend/steering/methods/registrations/ap20_method_stubs.py

@@ -2,7 +2,14 @@
 
 from __future__ import annotations
 
-from steering.methods.registrations._helpers import register_stub_method
+from steering.methods.registrations._helpers import (
+    SLICES_MATURITY,
+    SLICES_QUEUE,
+    SLICES_RECURRING,
+    SLICES_AGILE,
+    SLICES_OM_STANDARD,
+    register_stub_method,
+)
 
 
 def register() -> None:
@@ -11,40 +18,52 @@ def register() -> None:
         label="Reifegrad-Entwicklung",
         description="Stufen, Routinen, historische Entwicklung",
         next_action_strategy_key="maturity_progression",
+        data_slices=SLICES_MATURITY,
+        compatible_archetype_keys=frozenset({"initiative.maturity_journey"}),
     )
     register_stub_method(
         key="sequential_dependency",
         label="Sequenzielle Abhängigkeit",
         description="Graph-Pfade, kritischer Pfad (read model)",
         next_action_strategy_key="sequential_dependency",
+        data_slices=SLICES_OM_STANDARD,
+        compatible_archetype_keys="*",
     )
     register_stub_method(
         key="recurring_control",
         label="Rhythmus-Steuerung",
         description="Dauerprogramm, fällig/überfällig, Abweichungen",
         next_action_strategy_key="recurring_control",
+        data_slices=SLICES_RECURRING,
+        compatible_archetype_keys=frozenset({"initiative.recurring_program"}),
     )
     register_stub_method(
         key="queue_pull",
         label="Inbox / Queue",
         description="Pull oder Empfehlung aus Queue",
         next_action_strategy_key="queue_pull",
+        data_slices=SLICES_QUEUE,
+        compatible_archetype_keys="*",
     )
     register_stub_method(
         key="agile_iteration",
         label="Iterations-Zeitbox",
         description="Sprint/work_cycle Profil auf Product/Programm",
         next_action_strategy_key="agile_iteration",
+        data_slices=SLICES_AGILE,
+        compatible_archetype_keys="*",
     )
     register_stub_method(
         key="dispute_procedure",
         label="Verfahren / Konflikt",
         description="Reaktive Steuerung, Fristen, Entscheidungen",
         next_action_strategy_key="default",
+        compatible_archetype_keys=frozenset({"initiative.dispute_case"}),
     )
     register_stub_method(
         key="chapter_based_progression",
         label="Kapitel-Entwicklung",
         description="Inhaltliche Progression, Reviews",
         next_action_strategy_key="default",
+        compatible_archetype_keys=frozenset({"initiative.content_project"}),
     )

backend/steering/methods/registrations/generic_operating.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS
 from steering.methods.registry import MethodDefinition, get_method, register_method
+from steering.methods.registrations._helpers import SLICES_GENERIC
 
 
 def register() -> None:
@@ -17,5 +18,7 @@ def register() -> None:
             description="Standard-Lifecycle ohne methodenspezifische Spezialisierung",
             default_lifecycle_steps=STANDARD_LIFECYCLE_STEPS,
             next_action_strategy_key="default",
+            data_slices=SLICES_GENERIC,
+            compatible_archetype_keys="*",
         )
     )

backend/steering/methods/registrations/product_milestone_driven.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS
 from steering.methods.registry import MethodDefinition, get_method, register_method
+from steering.methods.registrations._helpers import SLICES_PRODUCT
 
 _PRODUCT_STEPS = (
     "intake",
@@ -32,5 +33,7 @@ def register() -> None:
             description="Meilenstein-orientierte Steuerung für Produkt- und Projektentwicklung",
             default_lifecycle_steps=_PRODUCT_STEPS,
             next_action_strategy_key="product_milestone_driven",
+            data_slices=SLICES_PRODUCT,
+            compatible_archetype_keys=frozenset({"initiative.product"}),
         )
     )

backend/steering/methods/registrations/program_delivery.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
-from steering.methods.registrations._helpers import register_product_like_method
+from steering.methods.registrations._helpers import (
+    SLICES_PROGRAM,
+    register_product_like_method,
+)
 
 
 def register() -> None:
@@ -11,4 +14,6 @@ def register() -> None:
         label="Programm (begrenzt)",
         description="Meilenstein- und gate-orientierte Steuerung mit Abschluss",
         next_action_strategy_key="program_delivery",
+        data_slices=SLICES_PROGRAM,
+        compatible_archetype_keys=frozenset({"initiative.program"}),
     )

backend/steering/methods/registry.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import Literal
 
 _METHODS: dict[str, "MethodDefinition"] = {}
 
@@ -15,6 +16,17 @@ class MethodDefinition:
     description: str
     default_lifecycle_steps: tuple[str, ...]
     next_action_strategy_key: str = "default"
+    data_slices: frozenset[str] = frozenset()
+    compatible_archetype_keys: frozenset[str] | Literal["*"] = "*"
+
+
+def method_compatible_with_archetype(
+    method: MethodDefinition, archetype_key: str
+) -> bool:
+    compat = method.compatible_archetype_keys
+    if compat == "*":
+        return True
+    return archetype_key in compat
 
 
 def register_method(defn: MethodDefinition) -> None:

backend/tests/test_ap23e_method_compatibility.py

@@ -0,0 +1,76 @@
+"""AP2.3e — Method compatibility validation."""
+
+from __future__ import annotations
+
+from tests.factories import provision_user_in_tenant
+from tests.test_initiatives_actions import _auth, _create_initiative, _login
+
+
+def test_update_method_queue_pull_on_linear_ok(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Linear Override",
+        archetype_key="initiative.linear_project",
+    )
+    initiative_id = created.json()["id"]
+
+    res = client.patch(
+        f"/api/steering/initiatives/{initiative_id}/context",
+        json={"method_key": "queue_pull"},
+        headers=_auth(token),
+    )
+    assert res.status_code == 200
+    assert res.json()["method_key"] == "queue_pull"
+
+
+def test_update_method_maturity_on_product_rejected(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Product Block",
+        archetype_key="initiative.product",
+    )
+    initiative_id = created.json()["id"]
+
+    res = client.patch(
+        f"/api/steering/initiatives/{initiative_id}/context",
+        json={"method_key": "maturity_progression"},
+        headers=_auth(token),
+    )
+    assert res.status_code == 400
+    assert "kompatibel" in res.json()["detail"].lower()
+
+
+def test_operating_context_reflects_method_slice_intersection(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Queue Linear",
+        archetype_key="initiative.linear_project",
+    )
+    initiative_id = created.json()["id"]
+
+    client.patch(
+        f"/api/steering/initiatives/{initiative_id}/context",
+        json={"method_key": "queue_pull"},
+        headers=_auth(token),
+    )
+
+    ctx = client.get(
+        f"/api/initiatives/{initiative_id}/operating-context",
+        headers=_auth(token),
+    )
+    assert ctx.status_code == 200
+    slices = ctx.json()["data_slices"]
+    assert "backlog" in slices
+    assert "work_cycles" not in slices