feat(steering): Kernel Spine v0.2 — Events, Slot-Provider, continuous_product

apply_steering_event an Action done, Gate verify/reopen und Blocker resolve; lifecycle-Ketten methodenaware via slot_map; continuous_product slot_map; pytest v0.2.

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

backend/services/actions.py

@@ -492,6 +492,20 @@ def update_action(
                 "to_status": status,
             },
         )
+        if status == "done":
+            from services.steering_events import emit_steering_event
+
+            emit_steering_event(
+                tenant_id=tenant_id,
+                initiative_id=str(existing["initiative_id"]),
+                event_kind="actor_result",
+                payload={
+                    "action_id": action_id,
+                    "from_status": old_status,
+                    "to_status": status,
+                },
+                user_id=user_id,
+            )
     return result
 
 

backend/services/operating_transitions.py

@@ -99,14 +99,13 @@ def after_blocker_status_change(
             user_id=user_id,
         )
         if initiative_id and user_id:
-            from steering.lifecycle.orchestrator import transition
+            from services.steering_events import emit_steering_event
 
-            transition(
-                None,
-                initiative_id,
-                "action_selection",
-                reason="blocker_resolved",
+            emit_steering_event(
                 tenant_id=tenant_id,
+                initiative_id=initiative_id,
+                event_kind="situational",
+                payload={"reason": "blocker_resolved", "blocker_id": blocker_id},
                 user_id=user_id,
             )
 

backend/services/roadmap.py

@@ -719,6 +719,27 @@ def verify_reached(
         )
         if transition:
             row["maturity_transition"] = transition
+    elif row.get("item_type") != "work_cycle":
+        from services.steering_events import emit_steering_event
+
+        steering = emit_steering_event(
+            tenant_id=tenant_id,
+            initiative_id=initiative_id,
+            event_kind="gate_verified",
+            payload={
+                "roadmap_item_id": item_id,
+                "item_type": row.get("item_type"),
+                "verify_reason": verify_reason,
+            },
+            user_id=user_id,
+        )
+        row["steering_event"] = {
+            "handled": steering.get("handled"),
+            "lifecycle_chain": steering.get("lifecycle_chain"),
+            "lifecycle_state": steering.get("evaluation", {})
+            .get("lifecycle", {})
+            .get("current_state"),
+        }
     return row
 
 
@@ -773,4 +794,22 @@ def reopen_roadmap_item(
             "to_status": "active",
         },
     )
+    if row.get("item_type") != "work_cycle":
+        from services.steering_events import emit_steering_event
+
+        steering = emit_steering_event(
+            tenant_id=tenant_id,
+            initiative_id=str(row["initiative_id"]),
+            event_kind="gate_reopened",
+            payload={
+                "roadmap_item_id": item_id,
+                "item_type": row.get("item_type"),
+                "reason": reason.strip() or None,
+            },
+            user_id=user_id,
+        )
+        row["steering_event"] = {
+            "handled": steering.get("handled"),
+            "lifecycle_chain": steering.get("lifecycle_chain"),
+        }
     return row

backend/services/steering_events.py

@@ -0,0 +1,26 @@
+"""Emit steering kernel events from OM mutations — Spine v0.2."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from steering.kernel.context import resolve_minimal_tenant_context
+from steering.kernel.events import apply_steering_event
+from steering.lifecycle.event_transitions import EventKind
+
+
+def emit_steering_event(
+    *,
+    tenant_id: str,
+    initiative_id: str,
+    event_kind: EventKind,
+    payload: dict[str, Any] | None = None,
+    user_id: Optional[str] = None,
+) -> dict[str, Any]:
+    ctx = resolve_minimal_tenant_context(tenant_id=tenant_id, user_id=user_id)
+    return apply_steering_event(
+        ctx,
+        initiative_id=initiative_id,
+        event_kind=event_kind,
+        payload=payload,
+    )

backend/steering/__init__.py

@@ -22,6 +22,9 @@ def bootstrap_steering() -> None:
     program_delivery.register()
     continuous_product.register()
     ap20_method_stubs.register()
+    from steering.lifecycle.slot_providers import register_default_slot_providers
+
+    register_default_slot_providers()
     from steering.methods.registry import validate_method_registry
 
     validate_method_registry()

backend/steering/kernel/__init__.py

@@ -1,7 +1,6 @@
-"""Steering Kernel Spine v0.1 — universal runtime entry for initiative steering."""
+"""Steering Kernel Spine — universal runtime entry for initiative steering."""
 
 from steering.kernel.evaluate import evaluate_steering
-from steering.kernel.events import apply_steering_event
 from steering.kernel.models import HorizonMarker, LifecycleContext, SteeringBinding, SteeringEvaluation
 
 __all__ = [
@@ -9,6 +8,5 @@ __all__ = [
     "LifecycleContext",
     "SteeringBinding",
     "SteeringEvaluation",
-    "apply_steering_event",
     "evaluate_steering",
 ]

backend/steering/kernel/context.py

@@ -0,0 +1,81 @@
+"""Minimal TenantContext for service-layer steering calls."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from db import get_connection
+from psycopg2.extras import RealDictCursor
+from rights_registry import load_grants_for_roles
+from tenant_context import TenantContext
+
+
+def resolve_minimal_tenant_context(
+    *,
+    tenant_id: str,
+    user_id: Optional[str] = None,
+) -> TenantContext:
+    """Build TenantContext for kernel calls from OM services (no HTTP session)."""
+    conn = get_connection()
+    try:
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                "SELECT id, slug, name FROM tenants WHERE id = %s",
+                (tenant_id,),
+            )
+            tenant = cur.fetchone()
+            if not tenant:
+                raise ValueError("Tenant nicht gefunden")
+
+            user_row = None
+            membership_role: str | None = None
+            actor_id: str | None = None
+            if user_id:
+                cur.execute(
+                    """
+                    SELECT u.id, u.email, u.display_name, u.portal_role, tm.tenant_role
+                    FROM users u
+                    LEFT JOIN tenant_memberships tm
+                      ON tm.user_id = u.id AND tm.tenant_id = %s AND tm.is_active = TRUE
+                    WHERE u.id = %s
+                    """,
+                    (tenant_id, user_id),
+                )
+                user_row = cur.fetchone()
+                if user_row:
+                    membership_role = user_row.get("tenant_role")
+                    cur.execute(
+                        """
+                        SELECT id FROM actors
+                        WHERE tenant_id = %s AND user_id = %s AND actor_type = 'human'
+                        LIMIT 1
+                        """,
+                        (tenant_id, user_id),
+                    )
+                    actor = cur.fetchone()
+                    actor_id = str(actor["id"]) if actor else None
+    finally:
+        conn.close()
+
+    portal_role = user_row["portal_role"] if user_row else "user"
+    caps = frozenset(
+        load_grants_for_roles(
+            portal_role=portal_role,
+            tenant_role=membership_role,
+        )
+    )
+
+    return TenantContext(
+        user_id=str(user_row["id"]) if user_row else "",
+        email=user_row["email"] if user_row else "",
+        display_name=user_row["display_name"] if user_row else "",
+        portal_role=portal_role,
+        tenant_id=str(tenant["id"]),
+        tenant_slug=tenant["slug"],
+        tenant_name=tenant["name"],
+        tenant_role=membership_role,
+        actor_id=actor_id,
+        actor_type="human" if actor_id else None,
+        session_token="steering-service",
+        capabilities=caps,
+    )

backend/steering/kernel/events.py

@@ -1,17 +1,20 @@
-"""Event ingress stub — dual trigger (Plan | Event) v0.1."""
+"""Event ingress — dual trigger (Plan | Event) v0.2."""
 
 from __future__ import annotations
 
-from typing import Any, Literal
+from typing import Any
 
+from steering.kernel.binding import resolve_primary_method_key
+from steering.kernel.evaluate import evaluate_steering
+from steering.lifecycle.event_transitions import EventKind
+from steering.lifecycle.orchestrator import transition
+from steering.lifecycle.slot_providers import (
+    register_default_slot_providers,
+    resolve_event_via_providers,
+)
 from tenant_context import TenantContext
 
-EventKind = Literal[
-    "actor_result",
-    "external_event",
-    "cadence_due",
-    "situational",
-]
+register_default_slot_providers()
 
 
 def apply_steering_event(
@@ -22,21 +25,48 @@ def apply_steering_event(
     payload: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """
-    Event-Schub in den Kernel (Stub v0.1).
+    Event-Schub in den Kernel.
 
-    Plan-dominante Methoden: Event wird protokolliert, Auswertung erfolgt
-    weiterhin über evaluate_steering (Plan-Zug). Volle Event-Pipeline folgt
-    für dispute_procedure, care_navigation, recurring_control.
+    Plan-dominante Methoden: Event aktualisiert lifecycle_state entlang slot_map,
+    danach Re-Evaluate über evaluate_steering (Plan-Zug bleibt führend für Next).
     """
-    from steering.kernel.evaluate import evaluate_steering
+    payload = payload or {}
+    primary_key = resolve_primary_method_key(ctx, initiative_id)
+    chain = resolve_event_via_providers(
+        primary_method_key=primary_key,
+        event_kind=event_kind,
+        payload=payload,
+    )
+
+    transitions: list[dict[str, Any]] = []
+    handled = bool(chain)
+
+    for slot_state in chain:
+        result = transition(
+            ctx,
+            initiative_id,
+            slot_state,
+            reason=f"steering_event:{event_kind}",
+            tenant_id=ctx.tenant_id,
+            user_id=ctx.user_id or None,
+        )
+        if not result.get("unchanged"):
+            transitions.append(result)
 
     evaluation = evaluate_steering(ctx, initiative_id=initiative_id)
+
     return {
         "accepted": True,
         "event_kind": event_kind,
         "initiative_id": initiative_id,
-        "handled": False,
-        "message": "Event ingress stub — re-evaluate via Plan-Zug",
-        "payload_keys": sorted((payload or {}).keys()),
+        "handled": handled,
+        "lifecycle_chain": list(chain),
+        "transitions": transitions,
+        "message": (
+            "Lifecycle aktualisiert — Re-Evaluate via Plan-Zug"
+            if handled
+            else "Event protokolliert — kein Lifecycle-Slot für diese Methode"
+        ),
+        "payload_keys": sorted(payload.keys()),
         "evaluation": evaluation.to_dict(),
     }

backend/steering/kernel/models.py

@@ -65,7 +65,7 @@ class SteeringEvaluation:
     lifecycle: LifecycleContext
     next_work: list[dict[str, Any]] = field(default_factory=list)
     attention: list[dict[str, Any]] = field(default_factory=list)
-    data_source: str = "steering_kernel_v0.1.1"
+    data_source: str = "steering_kernel_v0.2"
 
     def to_dict(self) -> dict[str, Any]:
         return {

backend/steering/lifecycle/event_transitions.py

@@ -0,0 +1,66 @@
+"""Lifecycle transition chains for kernel event ingress v0.2."""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from steering.lifecycle.slot_maps import SlotStatus
+
+EventKind = Literal[
+    "actor_result",
+    "gate_verified",
+    "gate_reopened",
+    "external_event",
+    "cadence_due",
+    "situational",
+]
+
+
+def _slot_accepts_event(status: SlotStatus, *, event_targets_slot: bool) -> bool:
+    if status == "active":
+        return True
+    if status == "optional" and event_targets_slot:
+        return True
+    return False
+
+
+def resolve_lifecycle_chain(
+    *,
+    primary_method_key: str,
+    event_kind: EventKind,
+    payload: dict[str, Any] | None = None,
+) -> tuple[str, ...]:
+    """
+    Ordered lifecycle slots to visit for an event (method-aware via slot_map).
+
+    Returns slot keys; caller persists the last applicable state.
+    """
+    from steering.kernel.lifecycle import resolve_method_slot_map
+
+    payload = payload or {}
+    slot_map = resolve_method_slot_map(primary_method_key)
+
+    if event_kind == "actor_result" and payload.get("to_status") == "done":
+        candidates = ("result_intake", "adaptation", "action_selection")
+        event_slot = "result_intake"
+    elif event_kind == "gate_verified":
+        candidates = ("validation", "adaptation", "action_selection")
+        event_slot = "validation"
+    elif event_kind == "gate_reopened":
+        candidates = ("adaptation", "planning", "action_selection")
+        event_slot = "adaptation"
+    elif event_kind == "situational" and payload.get("reason") == "blocker_resolved":
+        candidates = ("action_selection",)
+        event_slot = "action_selection"
+    else:
+        return ()
+
+    chain: list[str] = []
+    for step in candidates:
+        status = slot_map.get(step, "n/a")
+        if _slot_accepts_event(
+            status,  # type: ignore[arg-type]
+            event_targets_slot=(step == event_slot or step in ("adaptation", "action_selection")),
+        ):
+            chain.append(step)
+    return tuple(chain)

backend/steering/lifecycle/slot_maps.py

@@ -43,6 +43,23 @@ GENERIC_OPERATING_SLOTS: dict[str, SlotStatus] = {
 }
 
 
+# Spec-D continuous_product D5 (method_bind → method_selection)
+CONTINUOUS_PRODUCT_SLOTS: dict[str, SlotStatus] = {
+    "intake": "active",
+    "method_selection": "active",
+    "structure_setup": "active",
+    "planning": "active",
+    "action_selection": "active",
+    "assignment": "active",
+    "waiting": "active",
+    "result_intake": "active",
+    "validation": "optional",
+    "review": "optional",
+    "adaptation": "active",
+    "closure": "n/a",
+}
+
+
 def validate_slot_map(slot_map: dict[str, SlotStatus]) -> None:
     for step, status in slot_map.items():
         if step not in STANDARD_LIFECYCLE_STEPS:

backend/steering/lifecycle/slot_providers.py

@@ -0,0 +1,107 @@
+"""Lifecycle slot providers — minimal registry v0.2."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable, Literal
+
+from steering.lifecycle.event_transitions import EventKind
+
+SlotKey = Literal[
+    "result_intake",
+    "validation",
+    "adaptation",
+    "action_selection",
+    "planning",
+]
+
+TransitionResolver = Callable[
+    [str, EventKind, dict[str, Any] | None],
+    tuple[str, ...],
+]
+
+_PROVIDERS: dict[SlotKey, TransitionResolver] = {}
+
+
+@dataclass(frozen=True)
+class SlotProviderRegistration:
+    slot_key: SlotKey
+    event_kinds: frozenset[EventKind]
+
+
+def register_slot_provider(
+    slot_key: SlotKey,
+    *,
+    event_kinds: frozenset[EventKind],
+    resolver: TransitionResolver,
+) -> None:
+    _PROVIDERS[slot_key] = resolver
+    _REGISTRATIONS[slot_key] = SlotProviderRegistration(
+        slot_key=slot_key, event_kinds=event_kinds
+    )
+
+
+_REGISTRATIONS: dict[SlotKey, SlotProviderRegistration] = {}
+
+
+def list_slot_provider_registrations() -> tuple[SlotProviderRegistration, ...]:
+    return tuple(_REGISTRATIONS.values())
+
+
+def resolve_event_via_providers(
+    *,
+    primary_method_key: str,
+    event_kind: EventKind,
+    payload: dict[str, Any] | None = None,
+) -> tuple[str, ...]:
+    """First matching provider wins; fallback to default event_transitions."""
+    from steering.lifecycle.event_transitions import resolve_lifecycle_chain
+
+    payload = payload or {}
+    for slot_key, registration in _REGISTRATIONS.items():
+        if event_kind not in registration.event_kinds:
+            continue
+        resolver = _PROVIDERS.get(slot_key)
+        if resolver:
+            chain = resolver(primary_method_key, event_kind, payload)
+            if chain:
+                return chain
+    return resolve_lifecycle_chain(
+        primary_method_key=primary_method_key,
+        event_kind=event_kind,
+        payload=payload,
+    )
+
+
+def register_default_slot_providers() -> None:
+    from steering.lifecycle.event_transitions import resolve_lifecycle_chain
+
+    if _PROVIDERS:
+        return
+
+    def _default_resolver(
+        primary_method_key: str,
+        event_kind: EventKind,
+        payload: dict[str, Any] | None,
+    ) -> tuple[str, ...]:
+        return resolve_lifecycle_chain(
+            primary_method_key=primary_method_key,
+            event_kind=event_kind,
+            payload=payload,
+        )
+
+    register_slot_provider(
+        "result_intake",
+        event_kinds=frozenset({"actor_result"}),
+        resolver=_default_resolver,
+    )
+    register_slot_provider(
+        "validation",
+        event_kinds=frozenset({"gate_verified"}),
+        resolver=_default_resolver,
+    )
+    register_slot_provider(
+        "adaptation",
+        event_kinds=frozenset({"actor_result", "gate_verified", "gate_reopened"}),
+        resolver=_default_resolver,
+    )

backend/steering/methods/registrations/_helpers.py

@@ -132,6 +132,7 @@ def register_product_like_method(
     steering_elements: frozenset[str] | None = None,
     ui_features: frozenset[str] | None = None,
     graph_profile: GraphMethodProfile = STRICT,
+    slot_map: tuple[tuple[str, str], ...] | None = None,
 ) -> None:
     register_stub_method(
         key=key,
@@ -147,4 +148,5 @@ def register_product_like_method(
         or (ELEM_NEXT | frozenset({"gate_fulfillment", "work_cycle_scope"})),
         ui_features=ui_features or frozenset(),
         graph_profile=graph_profile,
+        slot_map=slot_map or (),
     )

backend/steering/methods/registrations/continuous_product.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from steering.graph.profiles import FULFILLMENT
+from steering.lifecycle.slot_maps import CONTINUOUS_PRODUCT_SLOTS, slot_map_to_tuple
 from steering.methods.registrations._helpers import (
     ELEM_NEXT,
     register_product_like_method,
@@ -21,4 +22,5 @@ def register() -> None:
         ),
         steering_elements=ELEM_NEXT
         | frozenset({"gate_fulfillment", "work_cycle_scope"}),
+        slot_map=slot_map_to_tuple(CONTINUOUS_PRODUCT_SLOTS),
     )

backend/tests/test_ap_kernel_spine.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from steering.kernel import apply_steering_event, evaluate_steering
+from steering.kernel.events import apply_steering_event
+from steering.kernel.evaluate import evaluate_steering
 from steering.kernel.binding import resolve_steering_binding
 from tests.factories import provision_user_in_tenant, tenant_context_from_user
 from tests.test_initiatives_actions import _auth, _create_initiative, _login
@@ -41,7 +42,7 @@ def test_evaluate_steering_returns_horizon_and_next(client):
 
     ctx = tenant_context_from_user(user)
     evaluation = evaluate_steering(ctx, initiative_id=initiative_id)
-    assert evaluation.data_source == "steering_kernel_v0.1.1"
+    assert evaluation.data_source == "steering_kernel_v0.2"
     assert evaluation.lifecycle.slot_map.get("action_selection") == "active"
     assert evaluation.lifecycle.slot_map.get("validation") == "active"
     assert evaluation.lifecycle.slot_map.get("review") == "optional"
@@ -76,7 +77,7 @@ def test_snapshot_routes_through_kernel_not_side_path(client):
         f"/api/initiatives/{initiative_id}/steering-snapshot",
         headers=_auth(token),
     )
-    assert snap.json()["steering_kernel"]["data_source"] == "steering_kernel_v0.1.1"
+    assert snap.json()["steering_kernel"]["data_source"] == "steering_kernel_v0.2"
     assert "lifecycle" in snap.json()["steering_kernel"]
     assert snap.json()["steering_kernel"]["lifecycle"]["slot_map"]["closure"] == "active"
 
@@ -98,8 +99,8 @@ def test_apply_steering_event_stub(client):
         ctx,
         initiative_id=initiative_id,
         event_kind="actor_result",
-        payload={"action_id": "test"},
+        payload={"action_id": "test", "to_status": "done"},
     )
     assert result["accepted"] is True
-    assert result["handled"] is False
-    assert result["evaluation"]["data_source"] == "steering_kernel_v0.1.1"
+    assert result["handled"] is True
+    assert result["evaluation"]["data_source"] == "steering_kernel_v0.2"

backend/tests/test_ap_kernel_spine_v02.py

@@ -0,0 +1,165 @@
+"""Steering Kernel Spine v0.2 — events + continuous_product."""
+
+from __future__ import annotations
+
+from tests.factories import provision_user_in_tenant
+from tests.test_initiatives_actions import _auth, _create_initiative, _login
+
+
+def test_action_done_emits_lifecycle_via_kernel(client):
+    user = provision_user_in_tenant(tenant_role="admin")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Event Action Done",
+        archetype_key="initiative.linear_project",
+        apply_starter_kit=False,
+    )
+    initiative_id = created.json()["id"]
+
+    action = client.post(
+        f"/api/initiatives/{initiative_id}/actions",
+        json={"title": "Erledigen", "status": "open"},
+        headers=_auth(token),
+    )
+    assert action.status_code == 201
+
+    done = client.patch(
+        f"/api/actions/{action.json()['id']}",
+        json={"status": "done"},
+        headers=_auth(token),
+    )
+    assert done.status_code == 200
+
+    ctx = client.get(
+        f"/api/initiatives/{initiative_id}/steering-context",
+        headers=_auth(token),
+    )
+    assert ctx.status_code == 200
+    assert ctx.json()["lifecycle_state"] == "action_selection"
+
+
+def test_gate_verify_emits_lifecycle_via_kernel(client):
+    user = provision_user_in_tenant(tenant_role="admin")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Event Gate",
+        archetype_key="initiative.linear_project",
+    )
+    initiative_id = created.json()["id"]
+
+    roadmap = client.get(
+        f"/api/initiatives/{initiative_id}/roadmap/items",
+        headers=_auth(token),
+    )
+    active_gate = next(i for i in roadmap.json() if i.get("status") == "active")
+
+    evidence = client.post(
+        f"/api/initiatives/{initiative_id}/evidence",
+        json={
+            "title": "Gate-Nachweis",
+            "roadmap_item_id": active_gate["id"],
+            "status": "accepted",
+        },
+        headers=_auth(token),
+    )
+    assert evidence.status_code == 201
+
+    verified = client.post(
+        f"/api/roadmap-items/{active_gate['id']}/verify-reached",
+        headers=_auth(token),
+    )
+    assert verified.status_code == 200
+    body = verified.json()
+    assert body.get("steering_event", {}).get("handled") is True
+    assert "validation" in (body.get("steering_event", {}).get("lifecycle_chain") or [])
+
+    ctx = client.get(
+        f"/api/initiatives/{initiative_id}/steering-context",
+        headers=_auth(token),
+    )
+    assert ctx.json()["lifecycle_state"] == "action_selection"
+
+
+def test_continuous_product_routes_through_spine(client):
+    user = provision_user_in_tenant(tenant_role="admin")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Product Spine",
+        archetype_key="initiative.product",
+    )
+    initiative_id = created.json()["id"]
+
+    snap = client.get(
+        f"/api/initiatives/{initiative_id}/steering-snapshot",
+        headers=_auth(token),
+    )
+    assert snap.status_code == 200
+    body = snap.json()
+    kernel = body["steering_kernel"]
+    assert kernel["primary_method_key"] == "continuous_product"
+    assert kernel["data_source"] == "steering_kernel_v0.2"
+    assert kernel["lifecycle"]["slot_map"]["closure"] == "n/a"
+    assert kernel["lifecycle"]["slot_map"]["planning"] == "active"
+    assert body.get("next_actions") is not None
+
+
+def test_product_action_done_lifecycle_chain(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Product Event",
+        archetype_key="initiative.product",
+    )
+    initiative_id = created.json()["id"]
+
+    action = client.post(
+        f"/api/initiatives/{initiative_id}/actions",
+        json={"title": "Fix Bug", "status": "open"},
+        headers=_auth(token),
+    )
+    assert action.status_code == 201
+
+    client.patch(
+        f"/api/actions/{action.json()['id']}",
+        json={"status": "done"},
+        headers=_auth(token),
+    )
+
+    ctx = client.get(
+        f"/api/initiatives/{initiative_id}/steering-context",
+        headers=_auth(token),
+    )
+    assert ctx.json()["lifecycle_state"] == "action_selection"
+
+
+def test_slot_providers_registered():
+    from steering.lifecycle.slot_providers import list_slot_provider_registrations
+
+    keys = {r.slot_key for r in list_slot_provider_registrations()}
+    assert "result_intake" in keys
+    assert "validation" in keys
+    assert "adaptation" in keys
+
+
+def test_resolve_lifecycle_chain_respects_slot_map():
+    from steering.lifecycle.event_transitions import resolve_lifecycle_chain
+
+    chain = resolve_lifecycle_chain(
+        primary_method_key="generic_operating",
+        event_kind="gate_verified",
+        payload={},
+    )
+    assert "validation" not in chain
+    assert chain == ("adaptation", "action_selection")

docs/architecture/ADP_Steering_Kernel_Spine_v0.1.md

@@ -88,6 +88,17 @@ Strategies bleiben **Plugin-Lieferanten** — werden nicht dupliziert, aber nur
 - `evaluate_steering()` liefert `lifecycle`: `current_state`, `slot_map`, `active_slots`
 - Snapshot-Feld `steering_kernel.lifecycle`
 
+### Phase v0.2 (2026-07-26) — Scope Lock
+
+| # | Lieferung |
+|---|-----------|
+| V1 | Event-Ingress: Action `done`, Gate verify/reopen, Blocker resolve → `apply_steering_event()` |
+| V2 | Lifecycle-Kette methodenaware via `slot_map` + minimale Slot-Provider |
+| V3 | `continuous_product` mit `slot_map` durch dieselbe Spine |
+| V4 | pytest `test_ap_kernel_spine_v02.py` |
+
+**Out of Scope v0.2:** Voller 12-Schritt-Orchestrator-Loop pro Snapshot; C1/Care event-voll; Frontend.
+
 ---
 
-*Bei Erweiterung: v0.2 Event-Pfad + Slot-Provider — v0.2 + PO-Freigabe.*
+*Bei Erweiterung: v0.3 weitere Methoden + Event-Methoden — PO-Freigabe.*

docs/product/Kairo_Implementation_Truth_Table_v0.1.md

@@ -86,7 +86,7 @@ Verhindert, dass Zielbild-Dokumente als Ist-Stand gelesen werden.
 | Initiative Steering Snapshot | ✓ | Actions + linked; Archetyp/Guidance AP2.0c |
 | Operating Context API | ✓ | AP2.3a: `GET …/operating-context`, `ui_profile_json` Migration 027 |
 | Archetyp-/Methoden-Plugin-Architektur | ✓ | AP2.3 A–F + AP2.4; Operating Context, Slice-Loader, Route Gating |
-| **Steering Kernel Spine** | ◐ | v0.1.1: evaluate_steering + slot_map (A2); Event-Ingress Stub |
+| **Steering Kernel Spine** | ◐ | v0.2: Events + slot_map + continuous_product; A2 Abnahme ✓ |
 | Steuerungselement-Registry | ✓ | AP2.4: `steering/elements/`, FE `steeringElementRegistry`, Methoden-Vertrag |
 | operating_phase | ✗ | entfernt AP1.2 |
 | signals (Snapshot) | ✓ | `snapshot_signals.py` |