AP2.4: Steuerungselement-Registry und Methoden-Vertrag.
Some checks failed
Deploy Development / deploy (push) Successful in 46s
Test Suite / pytest-backend (push) Failing after 3m17s
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 2s
Test Suite / compose-smoke (push) Has been skipped

MethodDefinition traegt steering_elements/ui_features/graph_profile; operating-context und FE-Registry ohne Archetyp-Ifs.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Lars 2026-07-25 11:27:59 +02:00
parent ac475ab92c
commit d3bf3c36e0
25 changed files with 837 additions and 118 deletions

View File

@ -73,7 +73,6 @@ GENERIC_UI_PROFILE: dict[str, Any] = {
*_COMMON_OM_SLICES[1:], *_COMMON_OM_SLICES[1:],
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
} }
INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = { INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
@ -99,10 +98,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {
"continuousProductWorkMode": True,
"steeringSnapshotOnWorkSprint": True,
},
}, },
"initiative.linear_project": { "initiative.linear_project": {
"processSteps": list(_LINEAR_PROCESS), "processSteps": list(_LINEAR_PROCESS),
@ -123,9 +118,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {
"criticalPathControl": True,
},
}, },
"initiative.maturity_journey": { "initiative.maturity_journey": {
"processSteps": list(_MATURITY_PROCESS), "processSteps": list(_MATURITY_PROCESS),
@ -145,7 +137,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
"initiative.program": { "initiative.program": {
"processSteps": list(_PROGRAM_PROCESS), "processSteps": list(_PROGRAM_PROCESS),
@ -167,7 +158,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
"initiative.support_queue": { "initiative.support_queue": {
"processSteps": list(_SUPPORT_QUEUE_PROCESS), "processSteps": list(_SUPPORT_QUEUE_PROCESS),
@ -183,7 +173,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
"initiative.recurring_program": { "initiative.recurring_program": {
"processSteps": list(_MATURITY_PROCESS), "processSteps": list(_MATURITY_PROCESS),
@ -200,7 +189,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
"initiative.content_project": { "initiative.content_project": {
"processSteps": list(_CONTENT_PROCESS), "processSteps": list(_CONTENT_PROCESS),
@ -219,7 +207,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
"initiative.dispute_case": { "initiative.dispute_case": {
"processSteps": list(_DISPUTE_PROCESS), "processSteps": list(_DISPUTE_PROCESS),
@ -237,7 +224,6 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
"steering_methods", "steering_methods",
], ],
"enabledModes": ["plan", "work", "control"], "enabledModes": ["plan", "work", "control"],
"uiFeatures": {},
}, },
} }
@ -245,6 +231,8 @@ INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
def resolve_ui_profile(archetype_key: str) -> dict[str, Any]: def resolve_ui_profile(archetype_key: str) -> dict[str, Any]:
"""UI-Profil aus Code-Seed; unbekannte Archetypen → Generic-Fallback.""" """UI-Profil aus Code-Seed; unbekannte Archetypen → Generic-Fallback."""
profile = INITIATIVE_UI_PROFILES.get(archetype_key) profile = INITIATIVE_UI_PROFILES.get(archetype_key)
if profile: resolved = dict(profile) if profile else dict(GENERIC_UI_PROFILE)
return dict(profile) slices = resolved.get("dataSlices") or []
return dict(GENERIC_UI_PROFILE) resolved["omCapabilities"] = list(resolved.get("omCapabilities") or slices)
resolved.pop("uiFeatures", None)
return resolved

View File

@ -1,4 +1,4 @@
"""Steering API — methods, profiles and context (AP1.1 / AP2.0).""" """Steering API — methods, profiles and context (AP1.1 / AP2.0 / AP2.4)."""
from __future__ import annotations from __future__ import annotations
@ -7,11 +7,16 @@ from typing import Optional
from capabilities import require_capability from capabilities import require_capability
from fastapi import APIRouter, Depends, HTTPException, Query from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from entity_archetypes.registry import resolve_default_method_key, resolve_ui_profile
from method_profiles.registry import list_method_profiles from method_profiles.registry import list_method_profiles
from services import initiatives as initiative_service from services import initiatives as initiative_service
from services import steering_context as sc_service from services import steering_context as sc_service
from steering.context import get_steering_context_dto from steering.context import get_steering_context_dto
from steering.methods.registry import list_methods from steering.methods.registry import (
list_compatible_methods,
list_methods,
method_to_dict,
)
from tenant_context import TenantContext from tenant_context import TenantContext
router = APIRouter(prefix="/api/steering", tags=["steering"]) router = APIRouter(prefix="/api/steering", tags=["steering"])
@ -23,20 +28,34 @@ class SteeringContextUpdateRequest(BaseModel):
clear_method_profile: bool = False clear_method_profile: bool = False
def _om_capabilities_for_archetype(archetype_key: str | None) -> frozenset[str] | None:
if not archetype_key:
return None
profile = resolve_ui_profile(archetype_key)
caps = profile.get("omCapabilities") or profile.get("dataSlices") or []
return frozenset(caps)
@router.get("/methods") @router.get("/methods")
def list_steering_methods( def list_steering_methods(
archetype_key: Optional[str] = Query(default=None),
ctx: TenantContext = Depends(require_capability("kairo.initiative.read")), ctx: TenantContext = Depends(require_capability("kairo.initiative.read")),
): ):
_ = ctx _ = ctx
om_caps = _om_capabilities_for_archetype(archetype_key)
if archetype_key:
methods = list_compatible_methods(archetype_key, om_capabilities=om_caps)
default_key = resolve_default_method_key(archetype_key)
else:
methods = list_methods()
default_key = None
return [ return [
{ {
"key": m.key, **method_to_dict(m),
"version": m.version, "is_default": m.key == default_key if default_key else False,
"label": m.label,
"description": m.description,
"next_action_strategy_key": m.next_action_strategy_key,
} }
for m in list_methods() for m in methods
] ]

View File

@ -1,4 +1,4 @@
"""Operating Context read model — AP2.3a.""" """Operating Context read model — AP2.3a / AP2.4."""
from __future__ import annotations from __future__ import annotations
@ -7,10 +7,20 @@ from typing import Any, Optional
from psycopg2.extras import RealDictCursor from psycopg2.extras import RealDictCursor
from db import get_connection from db import get_connection
from entity_archetypes.registry import get_ui_profile_json, resolve_ui_profile from entity_archetypes.registry import (
get_ui_profile_json,
resolve_default_method_key,
resolve_ui_profile,
)
from services.initiatives import get_initiative from services.initiatives import get_initiative
from services.steering_context import ensure_steering_context, get_steering_context from services.steering_context import ensure_steering_context, get_steering_context
from steering.methods.registry import get_method from steering.graph.profiles import LIGHT, graph_profile_as_dict
from steering.methods.registry import (
get_method,
list_compatible_methods,
method_to_dict,
ui_features_as_dict,
)
def _load_ui_profile_from_db(archetype_key: str) -> Optional[dict[str, Any]]: def _load_ui_profile_from_db(archetype_key: str) -> Optional[dict[str, Any]]:
@ -32,8 +42,6 @@ def _load_ui_profile_from_db(archetype_key: str) -> Optional[dict[str, Any]]:
if isinstance(profile, dict) and profile: if isinstance(profile, dict) and profile:
return dict(profile) return dict(profile)
return None return None
except Exception:
return None
finally: finally:
conn.close() conn.close()
@ -48,18 +56,22 @@ def _resolve_ui_profile(archetype_key: str) -> dict[str, Any]:
return resolve_ui_profile(archetype_key) return resolve_ui_profile(archetype_key)
def _resolve_om_capabilities(ui_profile: dict[str, Any]) -> frozenset[str]:
caps = ui_profile.get("omCapabilities") or ui_profile.get("dataSlices") or []
return frozenset(caps)
def _resolve_data_slices( def _resolve_data_slices(
*, *,
ui_profile: dict[str, Any], om_capabilities: frozenset[str],
method_key: str, method_key: str,
) -> list[str]: ) -> list[str]:
"""Schnittmenge Archetyp-Default ∩ Method-Slices (Phase E erweitert).""" """Schnittmenge Archetyp-OM ∩ Method-Slices."""
archetype_slices = ui_profile.get("dataSlices") or []
method = get_method(method_key) method = get_method(method_key)
if method and getattr(method, "data_slices", None): if method and method.data_slices:
method_slices = set(method.data_slices) method_slices = set(method.data_slices)
return [s for s in archetype_slices if s in method_slices] return [s for s in om_capabilities if s in method_slices]
return list(archetype_slices) return list(om_capabilities)
def _method_capabilities(method_key: str) -> dict[str, Any]: def _method_capabilities(method_key: str) -> dict[str, Any]:
@ -75,6 +87,41 @@ def _method_capabilities(method_key: str) -> dict[str, Any]:
} }
def _resolve_method_contract(method_key: str) -> dict[str, Any]:
method = get_method(method_key)
if not method:
return {
"steering_elements": [],
"ui_features": {},
"graph_profile": graph_profile_as_dict(LIGHT),
}
return {
"steering_elements": sorted(method.steering_elements),
"ui_features": ui_features_as_dict(method.ui_features),
"graph_profile": graph_profile_as_dict(method.graph_profile),
}
def _compatible_methods_payload(
archetype_key: str,
*,
om_capabilities: frozenset[str],
active_method_key: str,
) -> list[dict[str, Any]]:
default_key = resolve_default_method_key(archetype_key)
methods = list_compatible_methods(
archetype_key, om_capabilities=om_capabilities
)
return [
{
**method_to_dict(m, include_compatibility=False),
"is_default": m.key == default_key,
"is_active": m.key == active_method_key,
}
for m in methods
]
def get_operating_context( def get_operating_context(
*, tenant_id: str, initiative_id: str *, tenant_id: str, initiative_id: str
) -> Optional[dict[str, Any]]: ) -> Optional[dict[str, Any]]:
@ -90,12 +137,17 @@ def get_operating_context(
archetype_key = initiative["archetype_key"] archetype_key = initiative["archetype_key"]
ui_profile = _resolve_ui_profile(archetype_key) ui_profile = _resolve_ui_profile(archetype_key)
om_capabilities = _resolve_om_capabilities(ui_profile)
method_key = steering["method_key"] method_key = steering["method_key"]
metadata = steering.get("lifecycle_metadata") or {} metadata = steering.get("lifecycle_metadata") or {}
if isinstance(metadata, str): if isinstance(metadata, str):
metadata = {} metadata = {}
method_profile_key = metadata.get("method_profile_key") method_profile_key = metadata.get("method_profile_key")
data_slices = _resolve_data_slices(ui_profile=ui_profile, method_key=method_key) data_slices = _resolve_data_slices(
om_capabilities=om_capabilities,
method_key=method_key,
)
method_contract = _resolve_method_contract(method_key)
return { return {
"initiative_id": initiative_id, "initiative_id": initiative_id,
@ -103,6 +155,15 @@ def get_operating_context(
"method_key": method_key, "method_key": method_key,
"method_profile_key": method_profile_key, "method_profile_key": method_profile_key,
"ui_profile": ui_profile, "ui_profile": ui_profile,
"om_capabilities": sorted(om_capabilities),
"data_slices": data_slices, "data_slices": data_slices,
"method_capabilities": _method_capabilities(method_key), "method_capabilities": _method_capabilities(method_key),
"steering_elements": method_contract["steering_elements"],
"ui_features": method_contract["ui_features"],
"graph_profile": method_contract["graph_profile"],
"compatible_methods": _compatible_methods_payload(
archetype_key,
om_capabilities=om_capabilities,
active_method_key=method_key,
),
} }

View File

@ -16,6 +16,14 @@ DEFAULT_METHOD_KEY = "generic_operating"
DEFAULT_METHOD_VERSION = "0.1.0" DEFAULT_METHOD_VERSION = "0.1.0"
def _resolve_om_capabilities_for_initiative(initiative: dict[str, Any]) -> frozenset[str]:
from entity_archetypes.registry import resolve_ui_profile
profile = resolve_ui_profile(initiative["archetype_key"])
caps = profile.get("omCapabilities") or profile.get("dataSlices") or []
return frozenset(caps)
def _serialize_row(row: dict[str, Any]) -> dict[str, Any]: def _serialize_row(row: dict[str, Any]) -> dict[str, Any]:
result = dict(row) result = dict(row)
for key in ("id", "tenant_id", "initiative_id"): for key in ("id", "tenant_id", "initiative_id"):
@ -229,7 +237,11 @@ def update_method_key(
initiative = get_initiative(tenant_id=tenant_id, initiative_id=initiative_id) initiative = get_initiative(tenant_id=tenant_id, initiative_id=initiative_id)
if not initiative: if not initiative:
raise ValueError("Vorhaben nicht gefunden") raise ValueError("Vorhaben nicht gefunden")
if not method_compatible_with_archetype(method, initiative["archetype_key"]): if not method_compatible_with_archetype(
method,
initiative["archetype_key"],
om_capabilities=_resolve_om_capabilities_for_initiative(initiative),
):
raise ValueError( raise ValueError(
f"Methode {method_key} ist nicht kompatibel mit Archetyp " f"Methode {method_key} ist nicht kompatibel mit Archetyp "
f"{initiative['archetype_key']}" f"{initiative['archetype_key']}"

View File

@ -0,0 +1,15 @@
"""Steering element catalog — AP2.4."""
from steering.elements.registry import (
STEERING_ELEMENTS,
get_steering_element,
list_steering_elements,
validate_steering_element_keys,
)
__all__ = [
"STEERING_ELEMENTS",
"get_steering_element",
"list_steering_elements",
"validate_steering_element_keys",
]

View File

@ -0,0 +1,90 @@
"""Steering element catalog — AP2.4."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
ModeKey = Literal["plan", "work", "control"]
@dataclass(frozen=True)
class SteeringElementDefinition:
key: str
label: str
description: str
modes: tuple[ModeKey, ...]
STEERING_ELEMENTS: dict[str, SteeringElementDefinition] = {
"next_action_primary": SteeringElementDefinition(
key="next_action_primary",
label="Nächster Schritt",
description="Next-Action-Widget als primäre Steuerungsansicht",
modes=("control", "work"),
),
"critical_path": SteeringElementDefinition(
key="critical_path",
label="Kritischer Pfad",
description="Execution-Graph und Pfad-Priorisierung",
modes=("control",),
),
"gate_fulfillment": SteeringElementDefinition(
key="gate_fulfillment",
label="Gate-Erfüllung",
description="Zielzustände, Verify und Fortschritt",
modes=("control", "plan"),
),
"work_cycle_scope": SteeringElementDefinition(
key="work_cycle_scope",
label="Sprint-Zeitbox",
description="Aktive work_cycle und Sprint-Backlog",
modes=("work", "plan"),
),
"recurring_rhythm": SteeringElementDefinition(
key="recurring_rhythm",
label="Rhythmen",
description="RecurringElement und Journey-Fokus",
modes=("control",),
),
"queue_inbox": SteeringElementDefinition(
key="queue_inbox",
label="Queue / Eingang",
description="Pull-Steuerung aus Backlog/Queue",
modes=("plan", "work"),
),
"maturity_stage": SteeringElementDefinition(
key="maturity_stage",
label="Reifegrad-Stufen",
description="Stufen und Routinen in Plan/Kontrolle",
modes=("plan", "control"),
),
"chapter_progression": SteeringElementDefinition(
key="chapter_progression",
label="Kapitel-Fortschritt",
description="Kapitel-basierte Progression",
modes=("plan", "control"),
),
"dispute_timeline": SteeringElementDefinition(
key="dispute_timeline",
label="Verfahren / Fristen",
description="Fristen, Entscheidungen, Verfahrensstrang",
modes=("control",),
),
}
def get_steering_element(key: str) -> SteeringElementDefinition | None:
return STEERING_ELEMENTS.get(key)
def list_steering_elements() -> tuple[SteeringElementDefinition, ...]:
return tuple(STEERING_ELEMENTS.values())
def validate_steering_element_keys(keys: frozenset[str]) -> None:
unknown = [k for k in keys if k not in STEERING_ELEMENTS]
if unknown:
raise ValueError(
f"Unbekannte steering_elements: {', '.join(sorted(unknown))}"
)

View File

@ -1,4 +1,4 @@
"""Method graph profiles — AP1.4e.""" """Method graph profiles — AP1.4e / AP2.4 (via MethodDefinition)."""
from __future__ import annotations from __future__ import annotations
@ -19,25 +19,20 @@ _STRICT = GraphMethodProfile(enforce_gate_blocking=True, emphasize_fulfillment=F
_FULFILLMENT = GraphMethodProfile(enforce_gate_blocking=False, emphasize_fulfillment=True) _FULFILLMENT = GraphMethodProfile(enforce_gate_blocking=False, emphasize_fulfillment=True)
_LIGHT = GraphMethodProfile(enforce_gate_blocking=False, emphasize_fulfillment=False) _LIGHT = GraphMethodProfile(enforce_gate_blocking=False, emphasize_fulfillment=False)
METHOD_GRAPH_PROFILES: dict[str, GraphMethodProfile] = { STRICT = _STRICT
"generic_operating": _LIGHT, FULFILLMENT = _FULFILLMENT
"queue_pull": _LIGHT, LIGHT = _LIGHT
"dispute_procedure": _LIGHT,
"continuous_product": _FULFILLMENT,
"maturity_progression": _FULFILLMENT,
"chapter_based_progression": _FULFILLMENT,
"product_milestone_driven": _STRICT,
"program_delivery": _STRICT,
"sequential_dependency": _STRICT,
"agile_iteration": _STRICT,
"recurring_control": _FULFILLMENT,
}
def get_graph_profile(method_key: str | None) -> GraphMethodProfile: def get_graph_profile(method_key: str | None) -> GraphMethodProfile:
if not method_key: if not method_key:
return METHOD_GRAPH_PROFILES[DEFAULT_METHOD_KEY] return LIGHT
return METHOD_GRAPH_PROFILES.get(method_key, _STRICT) from steering.methods.registry import get_method
method = get_method(method_key)
if method:
return method.graph_profile
return _STRICT
def enforce_gate_blocking_for_method(method_key: str | None) -> bool: def enforce_gate_blocking_for_method(method_key: str | None) -> bool:

View File

@ -1,9 +1,15 @@
"""Shared registration helper for AP2.0 method stubs — AP2.3e data_slices.""" """Shared registration helper for AP2.0 method stubs — AP2.3e / AP2.4."""
from __future__ import annotations from __future__ import annotations
from typing import Literal from typing import Literal
from steering.graph.profiles import (
FULFILLMENT,
GraphMethodProfile,
LIGHT,
STRICT,
)
from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS
from steering.methods.registry import MethodDefinition, get_method, register_method from steering.methods.registry import MethodDefinition, get_method, register_method
@ -74,6 +80,8 @@ SLICES_GENERIC = SLICES_OM_STANDARD
SLICES_AGILE = SLICES_PRODUCT SLICES_AGILE = SLICES_PRODUCT
ELEM_NEXT = frozenset({"next_action_primary"})
def register_stub_method( def register_stub_method(
*, *,
@ -84,6 +92,11 @@ def register_stub_method(
lifecycle_steps: tuple[str, ...] | None = None, lifecycle_steps: tuple[str, ...] | None = None,
data_slices: frozenset[str] | None = None, data_slices: frozenset[str] | None = None,
compatible_archetype_keys: frozenset[str] | Literal["*"] = "*", compatible_archetype_keys: frozenset[str] | Literal["*"] = "*",
steering_elements: frozenset[str] | None = None,
ui_features: frozenset[str] | None = None,
graph_profile: GraphMethodProfile = LIGHT,
method_role: Literal["primary", "modifier"] = "primary",
composes_with: frozenset[str] | None = None,
) -> None: ) -> None:
if get_method(key): if get_method(key):
return return
@ -97,6 +110,11 @@ def register_stub_method(
next_action_strategy_key=next_action_strategy_key, next_action_strategy_key=next_action_strategy_key,
data_slices=data_slices or SLICES_GENERIC, data_slices=data_slices or SLICES_GENERIC,
compatible_archetype_keys=compatible_archetype_keys, compatible_archetype_keys=compatible_archetype_keys,
steering_elements=steering_elements or ELEM_NEXT,
ui_features=ui_features or frozenset(),
graph_profile=graph_profile,
method_role=method_role,
composes_with=composes_with or frozenset(),
) )
) )
@ -109,6 +127,9 @@ def register_product_like_method(
next_action_strategy_key: str, next_action_strategy_key: str,
data_slices: frozenset[str] | None = None, data_slices: frozenset[str] | None = None,
compatible_archetype_keys: frozenset[str] | Literal["*"] | None = None, compatible_archetype_keys: frozenset[str] | Literal["*"] | None = None,
steering_elements: frozenset[str] | None = None,
ui_features: frozenset[str] | None = None,
graph_profile: GraphMethodProfile = STRICT,
) -> None: ) -> None:
register_stub_method( register_stub_method(
key=key, key=key,
@ -120,4 +141,8 @@ def register_product_like_method(
compatible_archetype_keys=compatible_archetype_keys compatible_archetype_keys=compatible_archetype_keys
if compatible_archetype_keys is not None if compatible_archetype_keys is not None
else frozenset({"initiative.product"}), else frozenset({"initiative.product"}),
steering_elements=steering_elements
or (ELEM_NEXT | frozenset({"gate_fulfillment", "work_cycle_scope"})),
ui_features=ui_features or frozenset(),
graph_profile=graph_profile,
) )

View File

@ -1,13 +1,15 @@
"""AP2.0a — remaining method stubs.""" """AP2.0a — remaining method stubs — AP2.4 steering elements."""
from __future__ import annotations from __future__ import annotations
from steering.graph.profiles import FULFILLMENT, LIGHT, STRICT
from steering.methods.registrations._helpers import ( from steering.methods.registrations._helpers import (
ELEM_NEXT,
SLICES_AGILE,
SLICES_MATURITY, SLICES_MATURITY,
SLICES_OM_STANDARD,
SLICES_QUEUE, SLICES_QUEUE,
SLICES_RECURRING, SLICES_RECURRING,
SLICES_AGILE,
SLICES_OM_STANDARD,
register_stub_method, register_stub_method,
) )
@ -20,6 +22,9 @@ def register() -> None:
next_action_strategy_key="maturity_progression", next_action_strategy_key="maturity_progression",
data_slices=SLICES_MATURITY, data_slices=SLICES_MATURITY,
compatible_archetype_keys=frozenset({"initiative.maturity_journey"}), compatible_archetype_keys=frozenset({"initiative.maturity_journey"}),
steering_elements=ELEM_NEXT
| frozenset({"maturity_stage", "recurring_rhythm"}),
graph_profile=FULFILLMENT,
) )
register_stub_method( register_stub_method(
key="sequential_dependency", key="sequential_dependency",
@ -28,6 +33,10 @@ def register() -> None:
next_action_strategy_key="sequential_dependency", next_action_strategy_key="sequential_dependency",
data_slices=SLICES_OM_STANDARD, data_slices=SLICES_OM_STANDARD,
compatible_archetype_keys="*", compatible_archetype_keys="*",
steering_elements=ELEM_NEXT
| frozenset({"critical_path", "gate_fulfillment"}),
ui_features=frozenset({"criticalPathControl"}),
graph_profile=STRICT,
) )
register_stub_method( register_stub_method(
key="recurring_control", key="recurring_control",
@ -36,6 +45,8 @@ def register() -> None:
next_action_strategy_key="recurring_control", next_action_strategy_key="recurring_control",
data_slices=SLICES_RECURRING, data_slices=SLICES_RECURRING,
compatible_archetype_keys=frozenset({"initiative.recurring_program"}), compatible_archetype_keys=frozenset({"initiative.recurring_program"}),
steering_elements=ELEM_NEXT | frozenset({"recurring_rhythm"}),
graph_profile=FULFILLMENT,
) )
register_stub_method( register_stub_method(
key="queue_pull", key="queue_pull",
@ -44,6 +55,8 @@ def register() -> None:
next_action_strategy_key="queue_pull", next_action_strategy_key="queue_pull",
data_slices=SLICES_QUEUE, data_slices=SLICES_QUEUE,
compatible_archetype_keys="*", compatible_archetype_keys="*",
steering_elements=ELEM_NEXT | frozenset({"queue_inbox"}),
graph_profile=LIGHT,
) )
register_stub_method( register_stub_method(
key="agile_iteration", key="agile_iteration",
@ -52,6 +65,11 @@ def register() -> None:
next_action_strategy_key="agile_iteration", next_action_strategy_key="agile_iteration",
data_slices=SLICES_AGILE, data_slices=SLICES_AGILE,
compatible_archetype_keys="*", compatible_archetype_keys="*",
steering_elements=ELEM_NEXT | frozenset({"work_cycle_scope"}),
ui_features=frozenset({"steeringSnapshotOnWorkSprint"}),
graph_profile=STRICT,
method_role="modifier",
composes_with=frozenset({"continuous_product", "program_delivery"}),
) )
register_stub_method( register_stub_method(
key="dispute_procedure", key="dispute_procedure",
@ -69,6 +87,8 @@ def register() -> None:
} }
), ),
compatible_archetype_keys=frozenset({"initiative.dispute_case"}), compatible_archetype_keys=frozenset({"initiative.dispute_case"}),
steering_elements=ELEM_NEXT | frozenset({"dispute_timeline"}),
graph_profile=LIGHT,
) )
register_stub_method( register_stub_method(
key="chapter_based_progression", key="chapter_based_progression",
@ -87,4 +107,7 @@ def register() -> None:
} }
), ),
compatible_archetype_keys=frozenset({"initiative.content_project"}), compatible_archetype_keys=frozenset({"initiative.content_project"}),
steering_elements=ELEM_NEXT
| frozenset({"chapter_progression", "gate_fulfillment"}),
graph_profile=FULFILLMENT,
) )

View File

@ -1,8 +1,12 @@
"""Built-in method: continuous_product — AP2.0a.""" """Built-in method: continuous_product — AP2.0a / AP2.4."""
from __future__ import annotations from __future__ import annotations
from steering.methods.registrations._helpers import register_product_like_method from steering.graph.profiles import FULFILLMENT
from steering.methods.registrations._helpers import (
ELEM_NEXT,
register_product_like_method,
)
def register() -> None: def register() -> None:
@ -11,4 +15,10 @@ def register() -> None:
label="Produkt (kontinuierlich)", label="Produkt (kontinuierlich)",
description="Kontinuierlicher Betrieb — Ist zuerst, Plan als Orientierung", description="Kontinuierlicher Betrieb — Ist zuerst, Plan als Orientierung",
next_action_strategy_key="continuous_product", next_action_strategy_key="continuous_product",
graph_profile=FULFILLMENT,
ui_features=frozenset(
{"continuousProductWorkMode", "steeringSnapshotOnWorkSprint"}
),
steering_elements=ELEM_NEXT
| frozenset({"gate_fulfillment", "work_cycle_scope"}),
) )

View File

@ -1,10 +1,11 @@
"""Built-in method: generic_operating.""" """Built-in method: generic_operating — AP2.4."""
from __future__ import annotations from __future__ import annotations
from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS
from steering.graph.profiles import LIGHT
from steering.methods.registrations._helpers import ELEM_NEXT, SLICES_GENERIC
from steering.methods.registry import MethodDefinition, get_method, register_method from steering.methods.registry import MethodDefinition, get_method, register_method
from steering.methods.registrations._helpers import SLICES_GENERIC
def register() -> None: def register() -> None:
@ -20,5 +21,7 @@ def register() -> None:
next_action_strategy_key="default", next_action_strategy_key="default",
data_slices=SLICES_GENERIC, data_slices=SLICES_GENERIC,
compatible_archetype_keys="*", compatible_archetype_keys="*",
steering_elements=ELEM_NEXT,
graph_profile=LIGHT,
) )
) )

View File

@ -1,10 +1,10 @@
"""Built-in method: product_milestone_driven — AP1.1.""" """Built-in method: product_milestone_driven — AP1.1 / AP2.4."""
from __future__ import annotations from __future__ import annotations
from steering.lifecycle.states import STANDARD_LIFECYCLE_STEPS from steering.graph.profiles import STRICT
from steering.methods.registrations._helpers import ELEM_NEXT, SLICES_PRODUCT
from steering.methods.registry import MethodDefinition, get_method, register_method from steering.methods.registry import MethodDefinition, get_method, register_method
from steering.methods.registrations._helpers import SLICES_PRODUCT
_PRODUCT_STEPS = ( _PRODUCT_STEPS = (
"intake", "intake",
@ -42,5 +42,8 @@ def register() -> None:
"initiative.program", "initiative.program",
} }
), ),
steering_elements=ELEM_NEXT
| frozenset({"gate_fulfillment", "work_cycle_scope"}),
graph_profile=STRICT,
) )
) )

View File

@ -1,8 +1,9 @@
"""Built-in method: program_delivery — AP2.0a.""" """Built-in method: program_delivery — AP2.0a / AP2.4."""
from __future__ import annotations from __future__ import annotations
from steering.methods.registrations._helpers import ( from steering.methods.registrations._helpers import (
ELEM_NEXT,
SLICES_PROGRAM, SLICES_PROGRAM,
register_product_like_method, register_product_like_method,
) )
@ -16,4 +17,6 @@ def register() -> None:
next_action_strategy_key="program_delivery", next_action_strategy_key="program_delivery",
data_slices=SLICES_PROGRAM, data_slices=SLICES_PROGRAM,
compatible_archetype_keys=frozenset({"initiative.program"}), compatible_archetype_keys=frozenset({"initiative.program"}),
steering_elements=ELEM_NEXT
| frozenset({"gate_fulfillment", "work_cycle_scope"}),
) )

View File

@ -1,9 +1,12 @@
"""Method Registry — built-in steering methods.""" """Method Registry — built-in steering methods — AP2.3e / AP2.4."""
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Literal from typing import Any, Literal
from steering.elements.registry import validate_steering_element_keys
from steering.graph.profiles import GraphMethodProfile, LIGHT
_METHODS: dict[str, "MethodDefinition"] = {} _METHODS: dict[str, "MethodDefinition"] = {}
@ -18,20 +21,51 @@ class MethodDefinition:
next_action_strategy_key: str = "default" next_action_strategy_key: str = "default"
data_slices: frozenset[str] = frozenset() data_slices: frozenset[str] = frozenset()
compatible_archetype_keys: frozenset[str] | Literal["*"] = "*" compatible_archetype_keys: frozenset[str] | Literal["*"] = "*"
steering_elements: frozenset[str] = frozenset()
ui_features: frozenset[str] = frozenset()
graph_profile: GraphMethodProfile = LIGHT
method_role: Literal["primary", "modifier"] = "primary"
composes_with: frozenset[str] = frozenset()
def ui_features_as_dict(features: frozenset[str]) -> dict[str, bool]:
return {key: True for key in sorted(features)}
def method_compatible_with_archetype( def method_compatible_with_archetype(
method: MethodDefinition, archetype_key: str method: MethodDefinition,
archetype_key: str,
*,
om_capabilities: frozenset[str] | None = None,
) -> bool: ) -> bool:
compat = method.compatible_archetype_keys compat = method.compatible_archetype_keys
if compat == "*": if compat != "*" and archetype_key not in compat:
return False
if om_capabilities is not None and method.data_slices:
if not method.data_slices.issubset(om_capabilities):
return False
return True return True
return archetype_key in compat
def list_compatible_methods(
archetype_key: str,
*,
om_capabilities: frozenset[str] | None = None,
) -> tuple[MethodDefinition, ...]:
return tuple(
m
for m in _METHODS.values()
if method_compatible_with_archetype(
m, archetype_key, om_capabilities=om_capabilities
)
)
def register_method(defn: MethodDefinition) -> None: def register_method(defn: MethodDefinition) -> None:
if defn.key in _METHODS: if defn.key in _METHODS:
raise ValueError(f"Method already registered: {defn.key}") raise ValueError(f"Method already registered: {defn.key}")
if defn.steering_elements:
validate_steering_element_keys(defn.steering_elements)
_METHODS[defn.key] = defn _METHODS[defn.key] = defn
@ -43,5 +77,34 @@ def list_methods() -> tuple[MethodDefinition, ...]:
return tuple(_METHODS.values()) return tuple(_METHODS.values())
def method_to_dict(
method: MethodDefinition,
*,
include_compatibility: bool = True,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"key": method.key,
"version": method.version,
"label": method.label,
"description": method.description,
"next_action_strategy_key": method.next_action_strategy_key,
"steering_elements": sorted(method.steering_elements),
"ui_features": ui_features_as_dict(method.ui_features),
"graph_profile": {
"enforce_gate_blocking": method.graph_profile.enforce_gate_blocking,
"emphasize_fulfillment": method.graph_profile.emphasize_fulfillment,
},
"method_role": method.method_role,
"composes_with": sorted(method.composes_with),
"data_slices": sorted(method.data_slices),
}
if include_compatibility:
compat = method.compatible_archetype_keys
payload["compatible_archetype_keys"] = (
"*" if compat == "*" else sorted(compat)
)
return payload
def clear_methods_for_tests() -> None: def clear_methods_for_tests() -> None:
_METHODS.clear() _METHODS.clear()

View File

@ -55,6 +55,7 @@ def test_operating_context_linear(client):
assert res.status_code == 200 assert res.status_code == 200
body = res.json() body = res.json()
assert body["method_key"] == "sequential_dependency" assert body["method_key"] == "sequential_dependency"
assert "critical_path" in body["steering_elements"]
assert body["ui_profile"]["planDefaultRoute"] == "/plan/gates" assert body["ui_profile"]["planDefaultRoute"] == "/plan/gates"
assert "work_cycles" not in body["data_slices"] assert "work_cycles" not in body["data_slices"]

View File

@ -0,0 +1,124 @@
"""AP2.4 — Steering elements and method contract."""
from __future__ import annotations
from steering.elements.registry import validate_steering_element_keys
from steering.methods.registry import get_method, list_compatible_methods, ui_features_as_dict
from steering.graph.profiles import get_graph_profile
def test_all_methods_have_valid_steering_elements():
from steering.methods.registry import list_methods
for method in list_methods():
validate_steering_element_keys(method.steering_elements)
def test_sequential_dependency_contract():
method = get_method("sequential_dependency")
assert method is not None
assert "critical_path" in method.steering_elements
assert "criticalPathControl" in method.ui_features
assert get_graph_profile("sequential_dependency").enforce_gate_blocking is True
def test_continuous_product_ui_features_from_method():
method = get_method("continuous_product")
assert method is not None
features = ui_features_as_dict(method.ui_features)
assert features["continuousProductWorkMode"] is True
assert features["steeringSnapshotOnWorkSprint"] is True
def test_list_compatible_methods_linear_includes_queue_pull():
methods = list_compatible_methods("initiative.linear_project")
keys = {m.key for m in methods}
assert "sequential_dependency" in keys
assert "queue_pull" in keys
def test_list_compatible_methods_product_excludes_maturity():
methods = list_compatible_methods("initiative.product")
keys = {m.key for m in methods}
assert "continuous_product" in keys
assert "maturity_progression" not in keys
def test_operating_context_includes_method_contract(client):
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login
user = provision_user_in_tenant(tenant_role="member")
token = _login(client, user)
created = _create_initiative(
client,
token,
title="Linear Contract",
archetype_key="initiative.linear_project",
)
initiative_id = created.json()["id"]
res = client.get(
f"/api/initiatives/{initiative_id}/operating-context",
headers=_auth(token),
)
assert res.status_code == 200
body = res.json()
assert "critical_path" in body["steering_elements"]
assert body["ui_features"].get("criticalPathControl") is True
assert "graph_profile" in body
assert body["compatible_methods"]
assert any(m["key"] == "sequential_dependency" for m in body["compatible_methods"])
def test_steering_methods_filtered_by_archetype(client):
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _login
user = provision_user_in_tenant(tenant_role="member")
token = _login(client, user)
res = client.get(
"/api/steering/methods",
params={"archetype_key": "initiative.product"},
headers=_auth(token),
)
assert res.status_code == 200
keys = {m["key"] for m in res.json()}
assert "continuous_product" in keys
assert "maturity_progression" not in keys
product = next(m for m in res.json() if m["key"] == "continuous_product")
assert "steering_elements" in product
assert product["is_default"] is True
def test_queue_pull_on_linear_changes_steering_elements(client):
from tests.factories import provision_user_in_tenant
from tests.test_initiatives_actions import _auth, _create_initiative, _login
user = provision_user_in_tenant(tenant_role="member")
token = _login(client, user)
created = _create_initiative(
client,
token,
title="Linear Queue",
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),
)
body = ctx.json()
assert "queue_inbox" in body["steering_elements"]
assert "critical_path" not in body["steering_elements"]
assert body["ui_features"].get("criticalPathControl") is not True

View File

@ -0,0 +1,228 @@
# ADP — Steuerungselemente & Methoden-Vertrag v0.1 (AP2.4)
**Status:** vorgeschlagen — Implementierung AP2.3g/AP2.4
**Stand:** 2026-07-25
**Bezug:** `ADP_Archetype_Method_Plugin_Architecture_v0.1.md` (AP2.3), `ADP_Archetype_and_Method_Catalog_v0.2.md`, `Kairo_Method_Design_Principles_v0.1.md`
**Ersetzt nicht:** AP2.3 AF; **vertieft** Schicht 1 (Methode) und schließt Restschuld AP2.3g
---
## 1. Problem
AP2.3 hat Archetyp (Schicht 2) und Methode (Schicht 1) technisch entkoppelt:
- `GET /operating-context` liefert `ui_profile` + `data_slices`
- `MethodDefinition` trägt `data_slices` + `compatible_archetype_keys`
**Offene Lücke:** Steuerungsverhalten und Control-UI hängen noch an verstreuten Stellen:
| Verhalten | Heute | Soll |
|-----------|-------|------|
| Kritischer Pfad | FE-Hack `METHOD_UI_FEATURES` + Archetyp-If | Methode `sequential_dependency` |
| Product ohne Sprint | Archetyp `uiFeatures` | Methode `continuous_product` |
| Graph-Strenge | `graph/profiles.py` separat | Methoden-Vertrag |
| Control-Panels | Page-Ifs | Steuerungselement-Registry |
| Methodenwahl UI | alle Methoden | nur kompatible pro Archetyp |
**PO-Anforderung:** Jeder Archetyp kann mit **unterschiedlichen Steuerungsmethoden** geführt werden. Methoden bringen **eigene Steuerungselemente** mit und müssen **explizit** mit Archetypen kompatibel sein — ohne React-Ifs und ohne Duplikat-Registry.
---
## 2. Leitentscheidung
### 2.1 Drei Registries, eine Auflösung
```text
Archetyp-Registry (Schicht 2) → IA-Hülle, OM-Fähigkeiten, Default-Methode
Methoden-Registry (Schicht 1) → Motor: Strategie, Slices, Elemente, Graph
Steuerungselement-Registry → Katalog benannter Control-/Work-Bausteine
GET /initiatives/:id/operating-context
→ resolve(archetype, method, profile?)
```
### 2.2 Auflösungsregeln (verbindlich)
| Feld | Quelle | Regel |
|------|--------|-------|
| `ui_profile` (Nav, Routen, Outline) | **Archetyp** | unverändert |
| `data_slices` | **Schnittmenge** | `archetype.om_capabilities ∩ method.data_slices` |
| `steering_elements` | **Methode** | nur aktive Methode |
| `ui_features` | **Methode** | Verhaltensflags Work/Control |
| `graph_profile` | **Methode** | Gate-Blocking, Fulfillment |
| `method_capabilities` | **Methode** | Strategie, Lifecycle |
| `compatible_methods` | **Registry** | gefiltert nach Archetyp |
**Archetyp ≠ Methode:** Default-Methode ist **Empfehlung**, nicht Lock. Nutzer mit `kairo.initiative.manage` darf jede **kompatible** Methode wählen.
### 2.3 Kompatibilität
```text
compatible(archetype, method) :=
archetype_key ∈ method.compatible_archetype_keys (oder "*")
AND method.data_slices ⊆ archetype.om_capabilities
```
`om_capabilities` im Archetyp-Profil entspricht fachlich den erlaubten OM-Slices (Fallback: `dataSlices` aus `ui_profile_json`).
### 2.4 Methoden-Rollen (Komposition)
| Rolle | Bedeutung | Beispiel |
|-------|-----------|----------|
| `primary` | Einziger Motor am `steering_context.method_key` | `continuous_product` |
| `modifier` | Ergänzt Primärmethode über Profil/Policy | `agile_iteration` (Sprint auf Product) |
**MVP:** ein `method_key` pro Vorhaben; `agile_iteration` als Modifier über `method_profile_key` / `work_cycle`-Policy — kein zweiter SteeringContext.
---
## 3. Steuerungselemente (Schicht 1b)
Ein **Steuerungselement** ist ein stabiler Key, den eine Methode aktiviert. Backend und Frontend registrieren sich darauf.
| Key | Bedeutung | Typische Methode(n) |
|-----|-----------|---------------------|
| `next_action_primary` | Next-Action-Widget als Steuerungsdefault | alle |
| `critical_path` | Kritischer Pfad / Execution-Graph | `sequential_dependency` |
| `gate_fulfillment` | Gate-Fortschritt, Verify-Semantik | `program_delivery`, `product_milestone_driven`, … |
| `work_cycle_scope` | Sprint-Zeitbox, Sprint-Backlog | `continuous_product`, `agile_iteration` |
| `recurring_rhythm` | Rhythmen / Journey-Fokus | `maturity_progression`, `recurring_control` |
| `queue_inbox` | Queue-Pull, Eingang prominent | `queue_pull` |
| `maturity_stage` | Reifegrad-Stufen | `maturity_progression` |
| `chapter_progression` | Kapitel-Fortschritt | `chapter_based_progression` |
| `dispute_timeline` | Fristen / Verfahren | `dispute_procedure` |
**Registry:** `backend/steering/elements/registry.py`
**Frontend:** `frontend/src/registry/steeringElementRegistry.js` — Element → Panel/Widget
**Regel:** Control-/Work-Seiten composen UI aus `operating_context.steering_elements`**keine** Archetyp-Ifs.
### 3.1 UI-Features (Verhaltensflags)
Zusätzlich zu Elementen (was gerendert wird) können Methoden **Verhaltensflags** setzen:
| Key | Bedeutung |
|-----|-----------|
| `criticalPathControl` | Next-Action-Copy / Pfad-Fokus |
| `continuousProductWorkMode` | Product-Ist ohne aktiven Sprint |
| `steeringSnapshotOnWorkSprint` | Snapshot lazy auf Work/Sprint |
Quelle: **nur Methode** (`MethodDefinition.ui_features`), nicht Archetyp.
---
## 4. Methoden-Vertrag (`MethodDefinition`)
Erweiterung gegenüber AP2.3e:
```python
@dataclass(frozen=True)
class MethodDefinition:
key: str
version: str
label: str
description: str
default_lifecycle_steps: tuple[str, ...]
next_action_strategy_key: str
data_slices: frozenset[str]
compatible_archetype_keys: frozenset[str] | Literal["*"]
steering_elements: frozenset[str] = frozenset()
ui_features: frozenset[str] = frozenset()
graph_profile: GraphMethodProfile = LIGHT
method_role: Literal["primary", "modifier"] = "primary"
composes_with: frozenset[str] = frozenset()
```
`GraphMethodProfile` wandert aus `graph/profiles.py` in den Methoden-Vertrag; `get_graph_profile(method_key)` liest aus Registry.
---
## 5. Operating Context (API-Erweiterung)
`GET /api/initiatives/{id}/operating-context` liefert zusätzlich:
```json
{
"steering_elements": ["critical_path", "next_action_primary"],
"ui_features": { "criticalPathControl": true },
"graph_profile": {
"enforce_gate_blocking": true,
"emphasize_fulfillment": false
},
"compatible_methods": [
{ "key": "sequential_dependency", "label": "…", "is_default": true }
]
}
```
`ui_profile.uiFeatures` am Archetyp wird **deprecated** (leer); Migration in Methoden-Registrations.
---
## 6. Steering API
`GET /api/steering/methods?archetype_key=initiative.linear_project`
- Filtert nach `method_compatible_with_archetype`
- Response enthält `steering_elements`, `ui_features`, `compatible_archetype_keys`
Methoden-Dropdown in Control zeigt nur kompatible Einträge.
---
## 7. Implementierungspakete
| Paket | Lieferung | Abhängigkeit |
|-------|-----------|--------------|
| **AP2.4a** | `steering/elements/registry.py`, erweiterte `MethodDefinition`, Registrations | AP2.3e |
| **AP2.4b** | `operating-context` + `/steering/methods` Filter | AP2.4a |
| **AP2.4c** | FE Resolver, `steeringElementRegistry`, Pages ohne Ifs | AP2.4b |
| **AP2.4d** | pytest + Vitest + Truth Table | AP2.4c |
Optional **AP2.4e:** explizites `omCapabilities` am Archetyp-Seed (statt `dataSlices`-Missbrauch).
---
## 8. Erweiterungs-Workflow
**Neue Methode `foo_bar`:**
1. `steering/methods/registrations/foo_bar.py` — Contract vollständig
2. Nur Keys aus `steering/elements/registry.py` referenzieren
3. Import in `steering/__init__.py`
4. pytest `test_method_foo_bar_contract.py`
5. Kein Core-Edit außer Registry-Dispatcher
**Neues Steuerungselement `my_panel`:**
1. Eintrag in `steering/elements/registry.py`
2. FE `steeringElementRegistry.js`
3. Methoden, die es brauchen, tragen Key in `steering_elements`
---
## 9. Nicht tun
- Keine Archetyp-Ifs für Control (`initiative.linear_project === …`)
- Keine parallele Graph-Profile-Map ohne Methoden-Bezug
- Keine Tenant-DSL für Methoden
- Keine zweite SteeringContext-Tabelle für Komposition
---
## 10. Abnahme
| Kriterium | Test |
|-----------|------|
| Operating Context liefert `steering_elements` + `ui_features` aus Methode | pytest AP2.4 |
| Linear + `queue_pull` → keine `critical_path`, kleinere Slices | pytest AP2.3e |
| `/steering/methods?archetype_key=…` filtert | pytest AP2.4 |
| Control rendert Critical Path nur bei Element | Vitest + manuell |
| Kein `METHOD_UI_FEATURES` im FE | Code-Review |
---
## 11. Bezug AP2.3
AP2.3 AF liefert Operating Context, Slice-Loader, Route Gating.
AP2.4 **schließt die methodenunabhängige Steuerungsschicht** — Voraussetzung für Archetyp-Welle 2 (B1, A3, D1) ohne Page-Patches.

View File

@ -85,7 +85,8 @@ Verhindert, dass Zielbild-Dokumente als Ist-Stand gelesen werden.
| AttentionItem | ◐ | | | AttentionItem | ◐ | |
| Initiative Steering Snapshot | ✓ | Actions + linked; Archetyp/Guidance AP2.0c | | Initiative Steering Snapshot | ✓ | Actions + linked; Archetyp/Guidance AP2.0c |
| Operating Context API | ✓ | AP2.3a: `GET …/operating-context`, `ui_profile_json` Migration 027 | | Operating Context API | ✓ | AP2.3a: `GET …/operating-context`, `ui_profile_json` Migration 027 |
| Archetyp-/Methoden-Plugin-Architektur | ◐ | AP2.3 AE ✓; AP2.3f Restschuld (Resolver, Route-Gating, uiFeatures, slice reload) ◐ | | Archetyp-/Methoden-Plugin-Architektur | ◐ | AP2.3 AF ✓; AP2.4 Steuerungselemente + Methoden-Vertrag ◐ |
| Steuerungselement-Registry | ◐ | AP2.4: `steering/elements/`, FE `steeringElementRegistry` |
| operating_phase | ✗ | entfernt AP1.2 | | operating_phase | ✗ | entfernt AP1.2 |
| signals (Snapshot) | ✓ | `snapshot_signals.py` | | signals (Snapshot) | ✓ | `snapshot_signals.py` |
| Graph Read Models (blocked/ready) | ◐ | AP1.4d/e; Join/OR AP1.15d deferred | | Graph Read Models (blocked/ready) | ◐ | AP1.4d/e; Join/OR AP1.15d deferred |

View File

@ -1,7 +1,10 @@
import { apiFetch } from './client.js' import { apiFetch } from './client.js'
export function listSteeringMethods() { export function listSteeringMethods(archetypeKey = null) {
return apiFetch('/api/steering/methods') const qs = archetypeKey
? `?archetype_key=${encodeURIComponent(archetypeKey)}`
: ''
return apiFetch(`/api/steering/methods${qs}`)
} }
export function updateInitiativeSteeringMethod(initiativeId, methodKey) { export function updateInitiativeSteeringMethod(initiativeId, methodKey) {

View File

@ -998,6 +998,7 @@ export function InitiativeOperationsProvider({ children, initiativeId: initiativ
) )
const uiFeatures = operatingProfile.uiFeatures || {} const uiFeatures = operatingProfile.uiFeatures || {}
const steeringElements = operatingProfile.steeringElements || []
const value = { const value = {
initiativeId: id, initiativeId: id,
@ -1031,6 +1032,7 @@ export function InitiativeOperationsProvider({ children, initiativeId: initiativ
operatingContext, operatingContext,
dataSlices: operatingContext?.data_slices || [], dataSlices: operatingContext?.data_slices || [],
uiFeatures, uiFeatures,
steeringElements,
operatingProfile, operatingProfile,
methodBusy, methodBusy,
loading, loading,

View File

@ -4,11 +4,14 @@ import { useInitiativeOperations } from '../../context/InitiativeOperationsConte
import { SteeringSnapshotPanel } from '../../components/SteeringSnapshotPanel.jsx' import { SteeringSnapshotPanel } from '../../components/SteeringSnapshotPanel.jsx'
import { CriticalPathPanel } from '../../components/CriticalPathPanel.jsx' import { CriticalPathPanel } from '../../components/CriticalPathPanel.jsx'
import { NextActionWidget } from '../../widgets/NextActionWidget.jsx' import { NextActionWidget } from '../../widgets/NextActionWidget.jsx'
import {
hasSteeringElement,
steeringElementUi,
} from '../../registry/steeringElementRegistry.js'
export function InitiativeOverviewPage() { export function InitiativeOverviewPage() {
const { const {
initiativeId, initiativeId,
initiative,
actions, actions,
error, error,
steeringSnapshot, steeringSnapshot,
@ -18,11 +21,12 @@ export function InitiativeOverviewPage() {
methodBusy, methodBusy,
capabilities, capabilities,
handleMethodChange, handleMethodChange,
uiFeatures, steeringElements,
} = useInitiativeOperations() } = useInitiativeOperations()
const counts = steeringSnapshot?.counts || {} const counts = steeringSnapshot?.counts || {}
const showCriticalPath = Boolean(uiFeatures.criticalPathControl) const showCriticalPath = hasSteeringElement(steeringElements, 'critical_path')
const criticalPathUi = steeringElementUi(steeringElements, 'critical_path')
return ( return (
<> <>
@ -51,16 +55,8 @@ export function InitiativeOverviewPage() {
items={steeringSnapshot?.next_actions} items={steeringSnapshot?.next_actions}
loading={steeringSnapshotLoading} loading={steeringSnapshotLoading}
embedded embedded
title={ title={criticalPathUi?.nextActionTitle}
showCriticalPath subtitle={criticalPathUi?.nextActionSubtitle}
? 'Nächster Schritt am kritischen Pfad'
: undefined
}
subtitle={
showCriticalPath
? 'Empfehlung aus sequential_dependency — bereite Arbeitspakete zuerst.'
: undefined
}
/> />
)} )}

View File

@ -36,7 +36,10 @@ export const SLICE_LOADERS = {
workCycles: await listInitiativeWorkCycles(initiativeId), workCycles: await listInitiativeWorkCycles(initiativeId),
activeWorkCycle: await getActiveWorkCycle(initiativeId).catch(() => null), activeWorkCycle: await getActiveWorkCycle(initiativeId).catch(() => null),
}), }),
steering_methods: () => listSteeringMethods(), steering_methods: async (initiativeId) => {
const initiative = await getInitiative(initiativeId)
return listSteeringMethods(initiative?.archetype_key || null)
},
steering_snapshot: (initiativeId) => getInitiativeSteeringSnapshot(initiativeId), steering_snapshot: (initiativeId) => getInitiativeSteeringSnapshot(initiativeId),
} }

View File

@ -1,5 +1,5 @@
/** /**
* AP2.3b/f Operating Profile Resolver (API-first, methodUiDefaults nur als Fallback). * AP2.3b/f / AP2.4 Operating Profile Resolver (API-first, methodUiDefaults nur als Fallback).
*/ */
import { PLAN_OUTLINE_NODES } from '../plan/planOutlineNodes.js' import { PLAN_OUTLINE_NODES } from '../plan/planOutlineNodes.js'
import { WORK_NAV_ITEMS } from '../config/workNav.js' import { WORK_NAV_ITEMS } from '../config/workNav.js'
@ -20,9 +20,14 @@ import {
* archetype_key?: string | null, * archetype_key?: string | null,
* method_key?: string | null, * method_key?: string | null,
* method_profile_key?: string | null, * method_profile_key?: string | null,
* ui_profile?: Partial<MethodUiDefaults & { uiFeatures?: Record<string, boolean>, workDefaultRouteWithoutActiveWorkCycle?: string }> | null, * ui_profile?: Partial<MethodUiDefaults & { workDefaultRouteWithoutActiveWorkCycle?: string }> | null,
* data_slices?: string[], * data_slices?: string[],
* om_capabilities?: string[],
* method_capabilities?: object, * method_capabilities?: object,
* steering_elements?: string[],
* ui_features?: Record<string, boolean>,
* graph_profile?: { enforce_gate_blocking?: boolean, emphasize_fulfillment?: boolean },
* compatible_methods?: object[],
* }} OperatingContextResponse * }} OperatingContextResponse
* @typedef {{ * @typedef {{
* archetypeKey?: string | null, * archetypeKey?: string | null,
@ -40,20 +45,21 @@ const EMPTY_PROFILE = {
planOutlineKeys: null, planOutlineKeys: null,
workNavKeys: null, workNavKeys: null,
uiFeatures: {}, uiFeatures: {},
steeringElements: [],
} }
/** Methoden-Features ergänzen Archetyp-Profil (bis MethodDefinition ui_features trägt). */ function resolveUiFeatures(context, profileFeatures) {
const METHOD_UI_FEATURES = { if (context?.ui_features && Object.keys(context.ui_features).length) {
sequential_dependency: { criticalPathControl: true }, return { ...(context.ui_features || {}) }
continuous_product: { }
continuousProductWorkMode: true, return { ...(profileFeatures || {}) }
steeringSnapshotOnWorkSprint: true,
},
} }
function mergeUiFeatures(profileFeatures, methodKey) { function resolveSteeringElements(context) {
const methodFeatures = METHOD_UI_FEATURES[methodKey] || {} if (Array.isArray(context?.steering_elements)) {
return { ...methodFeatures, ...(profileFeatures || {}) } return [...context.steering_elements]
}
return []
} }
function resolveWorkDefaultRoute(profile, options = {}) { function resolveWorkDefaultRoute(profile, options = {}) {
@ -67,7 +73,6 @@ function resolveWorkDefaultRoute(profile, options = {}) {
/** /**
* @param {OperatingContextResponse | null | undefined} context * @param {OperatingContextResponse | null | undefined} context
* @param {{ hasActiveSprint?: boolean }} [options] * @param {{ hasActiveSprint?: boolean }} [options]
* @returns {MethodUiDefaults & { dataSlices: string[], uiFeatures: Record<string, boolean> }}
*/ */
export function resolveOperatingProfile(context, options = {}) { export function resolveOperatingProfile(context, options = {}) {
if (context?.ui_profile) { if (context?.ui_profile) {
@ -81,7 +86,8 @@ export function resolveOperatingProfile(context, options = {}) {
planOutlineKeys: profile.planOutlineKeys ?? null, planOutlineKeys: profile.planOutlineKeys ?? null,
workNavKeys: profile.workNavKeys ?? null, workNavKeys: profile.workNavKeys ?? null,
dataSlices: context.data_slices || profile.dataSlices || [], dataSlices: context.data_slices || profile.dataSlices || [],
uiFeatures: mergeUiFeatures(profile.uiFeatures, context?.method_key), uiFeatures: resolveUiFeatures(context, profile.uiFeatures),
steeringElements: resolveSteeringElements(context),
} }
} }
@ -93,13 +99,13 @@ export function resolveOperatingProfile(context, options = {}) {
return { return {
...fallback, ...fallback,
dataSlices: context?.data_slices || [], dataSlices: context?.data_slices || [],
uiFeatures: mergeUiFeatures({}, context?.method_key), uiFeatures: resolveUiFeatures(context, {}),
steeringElements: resolveSteeringElements(context),
} }
} }
/** /**
* @param {ResolveOperatingProfileInput} input * @param {ResolveOperatingProfileInput} input
* @returns {MethodUiDefaults & { dataSlices: string[], uiFeatures: Record<string, boolean> }}
*/ */
export function resolveOperatingProfileFromInput(input = {}) { export function resolveOperatingProfileFromInput(input = {}) {
if (input.operatingContext) { if (input.operatingContext) {
@ -112,7 +118,7 @@ export function resolveOperatingProfileFromInput(input = {}) {
methodKey: input.methodKey || null, methodKey: input.methodKey || null,
hasActiveSprint: input.hasActiveSprint, hasActiveSprint: input.hasActiveSprint,
}) })
return { ...fallback, dataSlices: [], uiFeatures: {} } return { ...fallback, dataSlices: [], uiFeatures: {}, steeringElements: [] }
} }
/** /**

View File

@ -5,6 +5,7 @@ import {
resolveOperatingProfileFromInput, resolveOperatingProfileFromInput,
resolveRedirectForDisallowedRoute, resolveRedirectForDisallowedRoute,
} from './resolveOperatingProfile.js' } from './resolveOperatingProfile.js'
import { hasSteeringElement } from './steeringElementRegistry.js'
const PRODUCT_CONTEXT = { const PRODUCT_CONTEXT = {
initiative_id: 'test-id', initiative_id: 'test-id',
@ -23,6 +24,11 @@ const PRODUCT_CONTEXT = {
workNavKeys: ['sprint', 'today', 'mine'], workNavKeys: ['sprint', 'today', 'mine'],
}, },
data_slices: ['backlog', 'actions', 'roadmap', 'work_cycles', 'projects'], data_slices: ['backlog', 'actions', 'roadmap', 'work_cycles', 'projects'],
steering_elements: ['next_action_primary', 'work_cycle_scope', 'gate_fulfillment'],
ui_features: {
continuousProductWorkMode: true,
steeringSnapshotOnWorkSprint: true,
},
} }
const SUPPORT_QUEUE_CONTEXT = { const SUPPORT_QUEUE_CONTEXT = {
@ -37,6 +43,8 @@ const SUPPORT_QUEUE_CONTEXT = {
workNavKeys: ['today', 'mine'], workNavKeys: ['today', 'mine'],
}, },
data_slices: ['backlog', 'actions', 'blockers'], data_slices: ['backlog', 'actions', 'blockers'],
steering_elements: ['next_action_primary', 'queue_inbox'],
ui_features: {},
} }
describe('resolveOperatingProfile', () => { describe('resolveOperatingProfile', () => {
@ -75,24 +83,28 @@ describe('resolveOperatingProfile', () => {
expect(isRouteAllowedForProfile('/plan/sprint', input)).toBe(true) expect(isRouteAllowedForProfile('/plan/sprint', input)).toBe(true)
}) })
it('liefert uiFeatures aus Archetyp-Profil', () => { it('liefert uiFeatures aus Methoden-Vertrag (API)', () => {
const profile = resolveOperatingProfile({ const profile = resolveOperatingProfile(PRODUCT_CONTEXT)
...PRODUCT_CONTEXT,
ui_profile: {
...PRODUCT_CONTEXT.ui_profile,
uiFeatures: { continuousProductWorkMode: true },
},
})
expect(profile.uiFeatures.continuousProductWorkMode).toBe(true) expect(profile.uiFeatures.continuousProductWorkMode).toBe(true)
expect(profile.uiFeatures.steeringSnapshotOnWorkSprint).toBe(true)
}) })
it('sequential_dependency ergänzt criticalPathControl', () => { it('liefert steeringElements aus Methoden-Vertrag (API)', () => {
const profile = resolveOperatingProfile({ const profile = resolveOperatingProfile({
archetype_key: 'initiative.generic', archetype_key: 'initiative.linear_project',
method_key: 'sequential_dependency', method_key: 'sequential_dependency',
ui_profile: { uiFeatures: {} }, ui_profile: PRODUCT_CONTEXT.ui_profile,
data_slices: ['actions'], steering_elements: ['critical_path', 'next_action_primary'],
ui_features: { criticalPathControl: true },
}) })
expect(hasSteeringElement(profile.steeringElements, 'critical_path')).toBe(true)
expect(profile.uiFeatures.criticalPathControl).toBe(true) expect(profile.uiFeatures.criticalPathControl).toBe(true)
}) })
}) })
describe('steeringElementRegistry', () => {
it('hasSteeringElement erkennt aktive Elemente', () => {
expect(hasSteeringElement(['critical_path'], 'critical_path')).toBe(true)
expect(hasSteeringElement(['queue_inbox'], 'critical_path')).toBe(false)
})
})

View File

@ -0,0 +1,33 @@
/**
* AP2.4 Steuerungselement-Registry (Element UI-Semantik).
*/
/** @type {Record<string, { nextActionTitle?: string, nextActionSubtitle?: string }>} */
export const STEERING_ELEMENT_UI = {
critical_path: {
nextActionTitle: 'Nächster Schritt am kritischen Pfad',
nextActionSubtitle:
'Empfehlung aus sequenzieller Abhängigkeit — bereite Arbeitspakete zuerst.',
},
queue_inbox: {
nextActionTitle: 'Nächster Schritt aus der Queue',
nextActionSubtitle: 'Pull-Empfehlung aus dem Eingang.',
},
}
/**
* @param {string[] | undefined | null} elements
* @param {string} key
*/
export function hasSteeringElement(elements, key) {
return Array.isArray(elements) && elements.includes(key)
}
/**
* @param {string[] | undefined | null} elements
* @param {string} key
*/
export function steeringElementUi(elements, key) {
if (!hasSteeringElement(elements, key)) return null
return STEERING_ELEMENT_UI[key] || null
}