AP2.3a: Operating Context API mit ui_profile_json und Archetyp-Sync.

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

backend/entity_archetypes/registry.py

@@ -7,6 +7,7 @@ from typing import Any, Optional
 from psycopg2.extras import RealDictCursor
 
 from db import get_connection
+from entity_archetypes.ui_profiles import INITIATIVE_UI_PROFILES, resolve_ui_profile
 
 # initiative suffix → default method (shared by initiative + project mirror)
 _METHOD_BY_SUFFIX: dict[str, str] = {
@@ -70,15 +71,18 @@ def _build_specs() -> tuple[dict[str, Any], ...]:
     specs: list[dict[str, Any]] = []
     for suffix, label, description in _INITIATIVE_SPEC_META:
         method = _METHOD_BY_SUFFIX[suffix]
+        initiative_key = f"initiative.{suffix}"
+        ui_profile = INITIATIVE_UI_PROFILES.get(initiative_key, {})
         specs.append(
             {
-                "key": f"initiative.{suffix}",
+                "key": initiative_key,
                 "entity_type": "initiative",
                 "label": label,
                 "description": description,
                 "is_system": True,
                 "default_method_key": method,
                 "default_method_profile_key": None,
+                "ui_profile_json": ui_profile,
             }
         )
         specs.append(
@@ -90,6 +94,7 @@ def _build_specs() -> tuple[dict[str, Any], ...]:
                 "is_system": True,
                 "default_method_key": method,
                 "default_method_profile_key": None,
+                "ui_profile_json": {},
             }
         )
     return tuple(specs)
@@ -189,6 +194,14 @@ def resolve_default_method_key(archetype_key: str) -> str:
     return spec.get("default_method_key", "generic_operating")
 
 
+def get_ui_profile_json(archetype_key: str) -> dict[str, Any]:
+    """UI-Profil aus Registry-Seed (DB-Sync-Quelle)."""
+    spec = _KEY_INDEX.get(archetype_key)
+    if spec and spec.get("ui_profile_json"):
+        return dict(spec["ui_profile_json"])
+    return resolve_ui_profile(archetype_key)
+
+
 def mirror_project_archetype_key(initiative_archetype_key: str) -> str:
     if initiative_archetype_key.startswith("initiative."):
         suffix = initiative_archetype_key[len("initiative.") :]

backend/entity_archetypes/ui_profiles.py

@@ -0,0 +1,181 @@
+"""UI-Profile für Initiative-Archetypen — migriert aus methodUiDefaults.js (AP2.3a)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+_PRODUCT_PROCESS = (
+    {"key": "inbox", "to": "/plan/inbox", "label": "Eingang", "mode": "plan"},
+    {"key": "sprint-plan", "to": "/plan/sprint", "label": "Sprint planen", "mode": "plan"},
+    {"key": "sprint-work", "to": "/work/sprint", "label": "Ausführen", "mode": "work"},
+    {"key": "control", "to": "/control/status", "label": "Kontrolle", "mode": "control"},
+)
+
+_LINEAR_PROCESS = (
+    {"key": "gates", "to": "/plan/gates", "label": "Zielzustände", "mode": "plan"},
+    {"key": "work", "to": "/work/today", "label": "Ausführen", "mode": "work"},
+    {"key": "control", "to": "/control/status", "label": "Kontrolle", "mode": "control"},
+)
+
+_MATURITY_PROCESS = (
+    {"key": "gates", "to": "/plan/gates", "label": "Stufen", "mode": "plan"},
+    {"key": "journey", "to": "/control/journey", "label": "Rhythmen", "mode": "control"},
+    {"key": "work", "to": "/work/today", "label": "Ausführen", "mode": "work"},
+    {"key": "control", "to": "/control/status", "label": "Kontrolle", "mode": "control"},
+)
+
+_PROGRAM_PROCESS = (
+    {"key": "gates", "to": "/plan/gates", "label": "Phasen", "mode": "plan"},
+    {"key": "inbox", "to": "/plan/inbox", "label": "Eingang", "mode": "plan"},
+    {"key": "work", "to": "/work/today", "label": "Ausführen", "mode": "work"},
+    {"key": "control", "to": "/control/status", "label": "Kontrolle", "mode": "control"},
+)
+
+_SUPPORT_QUEUE_PROCESS = (
+    {"key": "inbox", "to": "/plan/inbox", "label": "Eingang", "mode": "plan"},
+    {"key": "work", "to": "/work/today", "label": "Ausführen", "mode": "work"},
+    {"key": "control", "to": "/control/status", "label": "Kontrolle", "mode": "control"},
+)
+
+_COMMON_OM_SLICES = (
+    "actions",
+    "blockers",
+    "evidence",
+    "decisions",
+    "reviews",
+    "steering_methods",
+)
+
+GENERIC_UI_PROFILE: dict[str, Any] = {
+    "processSteps": [],
+    "planDefaultRoute": "/plan/profile",
+    "workDefaultRoute": "/work/today",
+    "controlDefaultRoute": "/control/status",
+    "planOutlineKeys": None,
+    "workNavKeys": None,
+    "dataSlices": ["actions", "backlog", "roadmap", "projects", "blockers", *_COMMON_OM_SLICES[1:]],
+    "enabledModes": ["plan", "work", "control"],
+}
+
+INITIATIVE_UI_PROFILES: dict[str, dict[str, Any]] = {
+    "initiative.product": {
+        "processSteps": list(_PRODUCT_PROCESS),
+        "planDefaultRoute": "/plan/inbox",
+        "workDefaultRoute": "/work/sprint",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "inbox", "sprint", "gates"],
+        "workNavKeys": ["sprint", "today", "mine"],
+        "dataSlices": [
+            "backlog",
+            "actions",
+            "roadmap",
+            "work_cycles",
+            "projects",
+            "blockers",
+            "evidence",
+            "decisions",
+            "reviews",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+    "initiative.linear_project": {
+        "processSteps": list(_LINEAR_PROCESS),
+        "planDefaultRoute": "/plan/gates",
+        "workDefaultRoute": "/work/today",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "structure", "gates", "inbox", "work"],
+        "workNavKeys": ["today", "mine", "sprint"],
+        "dataSlices": [
+            "backlog",
+            "actions",
+            "roadmap",
+            "projects",
+            "blockers",
+            "evidence",
+            "decisions",
+            "reviews",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+    "initiative.maturity_journey": {
+        "processSteps": list(_MATURITY_PROCESS),
+        "planDefaultRoute": "/plan/gates",
+        "workDefaultRoute": "/work/today",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "gates", "inbox", "work"],
+        "workNavKeys": ["today", "mine", "sprint"],
+        "dataSlices": [
+            "actions",
+            "roadmap",
+            "blockers",
+            "evidence",
+            "decisions",
+            "reviews",
+            "recurring",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+    "initiative.program": {
+        "processSteps": list(_PROGRAM_PROCESS),
+        "planDefaultRoute": "/plan/gates",
+        "workDefaultRoute": "/work/today",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "structure", "gates", "inbox", "sprint", "work"],
+        "workNavKeys": ["today", "mine", "sprint"],
+        "dataSlices": [
+            "backlog",
+            "actions",
+            "roadmap",
+            "projects",
+            "work_cycles",
+            "blockers",
+            "evidence",
+            "decisions",
+            "reviews",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+    "initiative.support_queue": {
+        "processSteps": list(_SUPPORT_QUEUE_PROCESS),
+        "planDefaultRoute": "/plan/inbox",
+        "workDefaultRoute": "/work/today",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "inbox", "work"],
+        "workNavKeys": ["today", "mine"],
+        "dataSlices": [
+            "backlog",
+            "actions",
+            "blockers",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+    "initiative.recurring_program": {
+        "processSteps": list(_MATURITY_PROCESS),
+        "planDefaultRoute": "/plan/gates",
+        "workDefaultRoute": "/work/today",
+        "controlDefaultRoute": "/control/status",
+        "planOutlineKeys": ["profile", "gates", "inbox", "work"],
+        "workNavKeys": ["today", "mine"],
+        "dataSlices": [
+            "actions",
+            "roadmap",
+            "recurring",
+            "blockers",
+            "steering_methods",
+        ],
+        "enabledModes": ["plan", "work", "control"],
+    },
+}
+
+
+def resolve_ui_profile(archetype_key: str) -> dict[str, Any]:
+    """UI-Profil aus Code-Seed; unbekannte Archetypen → Generic-Fallback."""
+    profile = INITIATIVE_UI_PROFILES.get(archetype_key)
+    if profile:
+        return dict(profile)
+    return dict(GENERIC_UI_PROFILE)

backend/entity_fields/sync.py

@@ -14,17 +14,20 @@ def sync_entity_field_registry_to_db() -> int:
     try:
         with conn.cursor() as cur:
             for archetype in ENTITY_ARCHETYPES:
+                ui_profile = archetype.get("ui_profile_json") or {}
                 cur.execute(
                     """
                     INSERT INTO entity_archetypes (
-                        archetype_key, entity_type, label, description, is_system
+                        archetype_key, entity_type, label, description, is_system,
+                        ui_profile_json
                     )
-                    VALUES (%s, %s, %s, %s, %s)
+                    VALUES (%s, %s, %s, %s, %s, %s::jsonb)
                     ON CONFLICT (archetype_key) DO UPDATE SET
                         entity_type = EXCLUDED.entity_type,
                         label = EXCLUDED.label,
                         description = EXCLUDED.description,
                         is_system = EXCLUDED.is_system,
+                        ui_profile_json = EXCLUDED.ui_profile_json,
                         updated_at = NOW()
                     """,
                     (
@@ -33,6 +36,7 @@ def sync_entity_field_registry_to_db() -> int:
                         archetype["label"],
                         archetype["description"],
                         archetype["is_system"],
+                        json.dumps(ui_profile),
                     ),
                 )
 

backend/migrations/027_entity_archetype_ui_profile.sql

@@ -0,0 +1,4 @@
+-- AP2.3a: UI-Profil pro Archetyp (Registry-Sync → Frontend Operating Context)
+
+ALTER TABLE entity_archetypes
+    ADD COLUMN IF NOT EXISTS ui_profile_json JSONB NOT NULL DEFAULT '{}'::jsonb;

backend/routers/initiatives.py

@@ -21,6 +21,7 @@ from services import blockers as blocker_service
 from services import decisions as decision_service
 from services import evidence as evidence_service
 from services import initiatives as initiative_service
+from services import operating_context as operating_context_service
 from services import milestones as milestone_service
 from services import recurring as recurring_service
 from services import reviews as review_service
@@ -213,6 +214,19 @@ def get_initiative_steering_context(
     return get_steering_context_dto(ctx, initiative_id=initiative_id)
 
 
+@router.get("/{initiative_id}/operating-context")
+def get_initiative_operating_context(
+    initiative_id: str,
+    ctx: TenantContext = Depends(require_capability("kairo.initiative.read")),
+):
+    context = operating_context_service.get_operating_context(
+        tenant_id=ctx.tenant_id, initiative_id=initiative_id
+    )
+    if not context:
+        raise HTTPException(status_code=404, detail="Vorhaben nicht gefunden")
+    return context
+
+
 def _require_initiative(*, tenant_id: str, initiative_id: str) -> dict:
     initiative = initiative_service.get_initiative(
         tenant_id=tenant_id, initiative_id=initiative_id

backend/services/operating_context.py

@@ -0,0 +1,108 @@
+"""Operating Context read model — AP2.3a."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from psycopg2.extras import RealDictCursor
+
+from db import get_connection
+from entity_archetypes.registry import get_ui_profile_json, resolve_ui_profile
+from services.initiatives import get_initiative
+from services.steering_context import ensure_steering_context, get_steering_context
+from steering.methods.registry import get_method
+
+
+def _load_ui_profile_from_db(archetype_key: str) -> Optional[dict[str, Any]]:
+    conn = get_connection()
+    try:
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT ui_profile_json
+                FROM entity_archetypes
+                WHERE archetype_key = %s
+                """,
+                (archetype_key,),
+            )
+            row = cur.fetchone()
+            if not row:
+                return None
+            profile = row.get("ui_profile_json")
+            if isinstance(profile, dict) and profile:
+                return dict(profile)
+            return None
+    except Exception:
+        return None
+    finally:
+        conn.close()
+
+
+def _resolve_ui_profile(archetype_key: str) -> dict[str, Any]:
+    db_profile = _load_ui_profile_from_db(archetype_key)
+    if db_profile:
+        return db_profile
+    seed_profile = get_ui_profile_json(archetype_key)
+    if seed_profile:
+        return seed_profile
+    return resolve_ui_profile(archetype_key)
+
+
+def _resolve_data_slices(
+    *,
+    ui_profile: dict[str, Any],
+    method_key: str,
+) -> list[str]:
+    """Schnittmenge Archetyp-Default ∩ Method-Slices (Phase E erweitert)."""
+    archetype_slices = ui_profile.get("dataSlices") or []
+    method = get_method(method_key)
+    if method and getattr(method, "data_slices", None):
+        method_slices = set(method.data_slices)
+        return [s for s in archetype_slices if s in method_slices]
+    return list(archetype_slices)
+
+
+def _method_capabilities(method_key: str) -> dict[str, Any]:
+    method = get_method(method_key)
+    if not method:
+        return {
+            "lifecycle_steps": [],
+            "next_action_strategy_key": "default",
+        }
+    return {
+        "lifecycle_steps": list(method.default_lifecycle_steps),
+        "next_action_strategy_key": method.next_action_strategy_key,
+    }
+
+
+def get_operating_context(
+    *, tenant_id: str, initiative_id: str
+) -> Optional[dict[str, Any]]:
+    initiative = get_initiative(tenant_id=tenant_id, initiative_id=initiative_id)
+    if not initiative:
+        return None
+
+    steering = get_steering_context(tenant_id=tenant_id, initiative_id=initiative_id)
+    if not steering:
+        steering = ensure_steering_context(
+            tenant_id=tenant_id, initiative_id=initiative_id
+        )
+
+    archetype_key = initiative["archetype_key"]
+    ui_profile = _resolve_ui_profile(archetype_key)
+    method_key = steering["method_key"]
+    metadata = steering.get("lifecycle_metadata") or {}
+    if isinstance(metadata, str):
+        metadata = {}
+    method_profile_key = metadata.get("method_profile_key")
+    data_slices = _resolve_data_slices(ui_profile=ui_profile, method_key=method_key)
+
+    return {
+        "initiative_id": initiative_id,
+        "archetype_key": archetype_key,
+        "method_key": method_key,
+        "method_profile_key": method_profile_key,
+        "ui_profile": ui_profile,
+        "data_slices": data_slices,
+        "method_capabilities": _method_capabilities(method_key),
+    }

backend/tests/test_ap23a_operating_context.py

@@ -0,0 +1,159 @@
+"""AP2.3a — Operating Context API."""
+
+from __future__ import annotations
+
+from auth import AUTH_HEADER
+from tests.factories import provision_user_in_tenant
+from tests.test_initiatives_actions import _auth, _create_initiative, _login
+
+
+def test_operating_context_product(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Product Alpha",
+        archetype_key="initiative.product",
+    )
+    assert created.status_code == 201
+    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 body["initiative_id"] == initiative_id
+    assert body["archetype_key"] == "initiative.product"
+    assert body["method_key"] == "continuous_product"
+    assert body["ui_profile"]["planDefaultRoute"] == "/plan/inbox"
+    assert body["ui_profile"]["workDefaultRoute"] == "/work/sprint"
+    assert "work_cycles" in body["data_slices"]
+    assert body["ui_profile"]["processSteps"]
+    assert body["method_capabilities"]["next_action_strategy_key"]
+
+
+def test_operating_context_linear(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Linear Proj",
+        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 body["method_key"] == "sequential_dependency"
+    assert body["ui_profile"]["planDefaultRoute"] == "/plan/gates"
+    assert "work_cycles" not in body["data_slices"]
+
+
+def test_operating_context_maturity(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Maturity",
+        archetype_key="initiative.maturity_journey",
+    )
+    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 body["method_key"] == "maturity_progression"
+    assert any(s["key"] == "journey" for s in body["ui_profile"]["processSteps"])
+
+
+def test_operating_context_program(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Program",
+        archetype_key="initiative.program",
+    )
+    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 body["method_key"] == "program_delivery"
+    assert body["ui_profile"]["planOutlineKeys"] is not None
+
+
+def test_operating_context_generic_fallback(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Dispute",
+        archetype_key="initiative.dispute_case",
+    )
+    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 body["ui_profile"]["planDefaultRoute"] == "/plan/profile"
+    assert body["ui_profile"]["processSteps"] == []
+
+
+def test_operating_context_support_queue_no_work_cycles(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    created = _create_initiative(
+        client,
+        token,
+        title="Support Queue",
+        archetype_key="initiative.support_queue",
+    )
+    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 body["method_key"] == "queue_pull"
+    assert "work_cycles" not in body["data_slices"]
+    assert "sprint" not in (body["ui_profile"].get("planOutlineKeys") or [])
+
+
+def test_operating_context_not_found(client):
+    user = provision_user_in_tenant(tenant_role="member")
+    token = _login(client, user)
+
+    res = client.get(
+        "/api/initiatives/00000000-0000-0000-0000-000000000099/operating-context",
+        headers=_auth(token),
+    )
+    assert res.status_code == 404