"""Plan vs. Ist overlay and diff read models (AP1.14).""" from __future__ import annotations from typing import Any, Optional TERMINAL_IST_STATUSES = frozenset({"reached", "moved", "discarded"}) IST_DEVIATION_CRITERION_STATUSES = frozenset({"waived", "deferred", "failed"}) def _item_def_key(item: dict[str, Any]) -> str: return str(item["id"]) def _dep_key(dep: dict[str, Any]) -> str: return ( f"{dep['from_item_id']}:{dep['to_item_id']}:" f"{dep.get('dependency_type', 'requires')}" ) def _crit_def_key(crit: dict[str, Any]) -> str: return str(crit["id"]) def _structural_diff( baseline: dict[str, Any], current: dict[str, Any] ) -> list[dict[str, Any]]: diffs: list[dict[str, Any]] = [] baseline_items = {_item_def_key(i): i for i in baseline.get("items", [])} current_items = {_item_def_key(i): i for i in current.get("items", [])} for item_id, item in current_items.items(): if item_id not in baseline_items: diffs.append( { "kind": "gate_added", "item_id": item_id, "title": item.get("title"), "message": "Gate im Plan neu seit letzter Revision", } ) else: base = baseline_items[item_id] if ( base.get("title") != item.get("title") or base.get("goal_description") != item.get("goal_description") or base.get("sort_order") != item.get("sort_order") ): diffs.append( { "kind": "gate_modified", "item_id": item_id, "title": item.get("title"), "message": "Gate-Definition geändert seit Revision", } ) for item_id, item in baseline_items.items(): if item_id not in current_items: diffs.append( { "kind": "gate_removed", "item_id": item_id, "title": item.get("title"), "message": "Gate aus Plan entfernt seit Revision", } ) baseline_deps = {_dep_key(d): d for d in baseline.get("dependencies", [])} current_deps = {_dep_key(d): d for d in current.get("dependencies", [])} for key, dep in current_deps.items(): if key not in baseline_deps: diffs.append( { "kind": "edge_added", "item_id": dep["from_item_id"], "related_item_id": dep["to_item_id"], "message": "Neue Kante seit Revision", } ) for key, dep in baseline_deps.items(): if key not in current_deps: diffs.append( { "kind": "edge_removed", "item_id": dep["from_item_id"], "related_item_id": dep["to_item_id"], "message": "Kante entfernt seit Revision", } ) baseline_crit = {_crit_def_key(c): c for c in baseline.get("criteria", [])} current_crit = {_crit_def_key(c): c for c in current.get("criteria", [])} for crit_id, crit in current_crit.items(): if crit_id not in baseline_crit: diffs.append( { "kind": "criterion_added", "item_id": crit["roadmap_item_id"], "criterion_id": crit_id, "title": crit.get("title"), "message": "Kriterium neu seit Revision", } ) elif baseline_crit[crit_id].get("title") != crit.get("title"): diffs.append( { "kind": "criterion_modified", "item_id": crit["roadmap_item_id"], "criterion_id": crit_id, "title": crit.get("title"), "message": "Kriterium geändert seit Revision", } ) for crit_id, crit in baseline_crit.items(): if crit_id not in current_crit: diffs.append( { "kind": "criterion_removed", "item_id": crit["roadmap_item_id"], "criterion_id": crit_id, "title": crit.get("title"), "message": "Kriterium entfernt seit Revision", } ) return diffs def _ist_deviation_diff( *, items: list[dict[str, Any]], criteria_by_item: dict[str, list[dict[str, Any]]], ) -> list[dict[str, Any]]: diffs: list[dict[str, Any]] = [] for item in items: item_id = str(item["id"]) status = item.get("status", "planned") if status in TERMINAL_IST_STATUSES: diffs.append( { "kind": "ist_gate_status", "item_id": item_id, "title": item.get("title"), "ist_status": status, "message": f"Ist-Status: {status}", } ) elif status == "at_risk": diffs.append( { "kind": "ist_gate_at_risk", "item_id": item_id, "title": item.get("title"), "ist_status": status, "message": "Gate als gefährdet markiert", } ) for crit in criteria_by_item.get(item_id, []): crit_status = crit.get("status", "open") if crit_status in IST_DEVIATION_CRITERION_STATUSES: diffs.append( { "kind": f"ist_criterion_{crit_status}", "item_id": item_id, "criterion_id": crit["id"], "title": crit.get("title"), "ist_status": crit_status, "message": f"Kriterium {crit_status}: {crit.get('title')}", } ) return diffs def build_plan_ist_view( *, tenant_id: str, initiative_id: str, snapshot: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: from services import roadmap as roadmap_service from services import roadmap_criteria as criteria_service from services.roadmap_plan_snapshot import capture_plan_payload from steering.graph.roadmap_engine import load_initiative_graph_state live_items = roadmap_service.list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) current_plan = capture_plan_payload(tenant_id=tenant_id, initiative_id=initiative_id) graph_state = load_initiative_graph_state( tenant_id=tenant_id, initiative_id=initiative_id ) baseline_plan = snapshot["plan_payload"] if snapshot else current_plan plan_source = "snapshot" if snapshot else "live" criteria_by_item: dict[str, list[dict[str, Any]]] = {} ist_items = [] for item in live_items: item_id = str(item["id"]) criteria = criteria_service.list_criteria_for_item( tenant_id=tenant_id, item_id=item_id ) criteria_by_item[item_id] = criteria graph_item = graph_state.get("items", {}).get(item_id, {}) ist_items.append( { "id": item_id, "title": item["title"], "status": item["status"], "fulfillment_ratio": graph_item.get("fulfillment_ratio"), "blocked": graph_item.get("blocked"), "ready": graph_item.get("ready"), "criteria": [ { "id": c["id"], "title": c["title"], "status": c["status"], "criterion_kind": c["criterion_kind"], } for c in criteria ], } ) diff: list[dict[str, Any]] = [] if snapshot: diff.extend(_structural_diff(baseline_plan, current_plan)) diff.extend( _ist_deviation_diff(items=live_items, criteria_by_item=criteria_by_item) ) latest_meta = None if snapshot: latest_meta = { "id": snapshot["id"], "revision_number": snapshot["revision_number"], "reason": snapshot.get("reason") or "", "created_at": snapshot.get("created_at"), } return { "initiative_id": initiative_id, "plan_source": plan_source, "latest_snapshot": latest_meta, "plan": baseline_plan, "current_plan": current_plan if snapshot else None, "ist": { "items": ist_items, "graph_state": { "method_key": graph_state.get("method_key"), "graph_profile": graph_state.get("graph_profile"), "initiative_fulfillment_ratio": graph_state.get( "initiative_fulfillment_ratio" ), }, }, "diff": diff, "diff_count": len(diff), } def load_plan_ist_view(*, tenant_id: str, initiative_id: str) -> dict[str, Any]: from services.initiatives import get_initiative from services.roadmap_plan_snapshot import list_plan_snapshots if not get_initiative(tenant_id=tenant_id, initiative_id=initiative_id): raise ValueError("Initiative nicht gefunden") snapshots = list_plan_snapshots( tenant_id=tenant_id, initiative_id=initiative_id, limit=1 ) snapshot = snapshots[0] if snapshots else None return build_plan_ist_view( tenant_id=tenant_id, initiative_id=initiative_id, snapshot=snapshot, )