"""A1 progression graph — multi-strand work gates, join switch nodes (PO-Lock §10).""" from __future__ import annotations from typing import Any, Optional from psycopg2.extras import RealDictCursor from db import get_connection WORK_GATE_TYPES = frozenset({"maturity_stage"}) JOIN_GATE_TYPE = "join_gate" ACTIVE_WORK_STATUSES = frozenset({"active", "at_risk"}) TERMINAL_STATUSES = frozenset({"reached", "moved", "discarded"}) def is_work_gate(item: dict[str, Any]) -> bool: return item.get("item_type") in WORK_GATE_TYPES def is_join_gate(item: dict[str, Any]) -> bool: return item.get("item_type") == JOIN_GATE_TYPE def _requires_deps(dependencies: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ dep for dep in dependencies if (dep.get("dependency_type") or "requires") == "requires" ] def join_input_gate_ids( join_id: str, dependencies: list[dict[str, Any]] ) -> list[str]: """Join requires input gates (from=join, to=input).""" return [ str(dep["to_item_id"]) for dep in _requires_deps(dependencies) if str(dep["from_item_id"]) == str(join_id) ] def join_output_gate_ids( join_id: str, dependencies: list[dict[str, Any]] ) -> list[str]: """Output gates require join (from=output, to=join).""" return [ str(dep["from_item_id"]) for dep in _requires_deps(dependencies) if str(dep["to_item_id"]) == str(join_id) ] def successor_work_gate_ids( gate_id: str, dependencies: list[dict[str, Any]] ) -> list[str]: """Work gates that require this gate (from=successor, to=gate).""" return [ str(dep["from_item_id"]) for dep in _requires_deps(dependencies) if str(dep["to_item_id"]) == str(gate_id) ] def list_active_work_gates( *, tenant_id: str, initiative_id: str ) -> list[dict[str, Any]]: from services.roadmap import list_roadmap_items_for_initiative items = list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) has_join_graph = any(is_join_gate(item) for item in items) active = [ item for item in items if is_work_gate(item) and item.get("status") in ACTIVE_WORK_STATUSES and (not has_join_graph or item.get("strand_project_id") is not None) ] active.sort(key=lambda item: (item.get("sort_order", 0), item.get("title", ""))) return active def compute_join_states( *, tenant_id: str, initiative_id: str, ) -> list[dict[str, Any]]: from services.projects import list_projects_for_initiative from services.roadmap import list_dependencies_for_initiative, list_roadmap_items_for_initiative items = list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) item_by_id = {str(item["id"]): item for item in items} dependencies = list_dependencies_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) projects = list_projects_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) project_title = {str(p["id"]): p.get("title") or "" for p in projects} joins = [item for item in items if is_join_gate(item)] states: list[dict[str, Any]] = [] for join in joins: join_id = str(join["id"]) input_ids = join_input_gate_ids(join_id, dependencies) output_ids = join_output_gate_ids(join_id, dependencies) pending: list[dict[str, Any]] = [] for input_id in input_ids: gate = item_by_id.get(input_id) if not gate: continue if gate.get("status") != "reached": strand_id = gate.get("strand_project_id") pending.append( { "gate_id": input_id, "title": gate.get("title") or input_id, "strand_project_id": strand_id, "strand_title": project_title.get(str(strand_id or ""), ""), "status": gate.get("status"), } ) states.append( { "join_id": join_id, "title": join.get("title") or "", "status": join.get("status"), "ready_to_switch": len(pending) == 0 and join.get("status") not in TERMINAL_STATUSES, "pending_inputs": pending, "output_gate_ids": output_ids, "outputs": [ { "gate_id": output_id, "title": (item_by_id.get(output_id) or {}).get("title") or output_id, "strand_project_id": (item_by_id.get(output_id) or {}).get( "strand_project_id" ), } for output_id in output_ids if output_id in item_by_id ], } ) return states def compute_progression_state( *, tenant_id: str, initiative_id: str ) -> dict[str, Any]: active_gates = list_active_work_gates( tenant_id=tenant_id, initiative_id=initiative_id ) return { "active_work_gates": [ { "id": gate["id"], "title": gate.get("title"), "strand_project_id": gate.get("strand_project_id"), "status": gate.get("status"), } for gate in active_gates ], "joins": compute_join_states( tenant_id=tenant_id, initiative_id=initiative_id ), } def enforce_single_active_work_gate_per_strand( cur, *, tenant_id: str, initiative_id: str, keep_active_id: str, strand_project_id: Optional[str], ) -> None: if strand_project_id: cur.execute( """ UPDATE roadmap_items ri SET status = 'planned', updated_at = NOW() FROM roadmaps r WHERE ri.roadmap_id = r.id AND ri.tenant_id = %s AND r.initiative_id = %s AND ri.item_type = 'maturity_stage' AND ri.status IN ('active', 'at_risk') AND ri.strand_project_id = %s AND ri.id != %s """, (tenant_id, initiative_id, strand_project_id, keep_active_id), ) return cur.execute( """ UPDATE roadmap_items ri SET status = 'planned', updated_at = NOW() FROM roadmaps r WHERE ri.roadmap_id = r.id AND ri.tenant_id = %s AND r.initiative_id = %s AND ri.item_type = 'maturity_stage' AND ri.status IN ('active', 'at_risk') AND ri.strand_project_id IS NULL AND ri.id != %s """, (tenant_id, initiative_id, keep_active_id), ) def _graph_ready( item_id: str, *, item_by_id: dict[str, dict[str, Any]], dependencies: list[dict[str, Any]], ) -> bool: for dep in _requires_deps(dependencies): if str(dep["from_item_id"]) != str(item_id): continue prereq = item_by_id.get(str(dep["to_item_id"])) if not prereq: continue if prereq.get("status") != "reached": return False return True def activate_work_gate( *, tenant_id: str, initiative_id: str, gate: dict[str, Any], user_id: Optional[str] = None, seed_practices: bool = True, ) -> dict[str, Any]: from services.maturity_practice import activate_stage_practice_set from services.roadmap import get_roadmap_item, update_roadmap_item gate_id = gate["id"] updated = update_roadmap_item( tenant_id=tenant_id, item_id=gate_id, status="active", user_id=user_id, ) if not updated: raise ValueError("Work-Gate konnte nicht aktiviert werden") practice_result = None if seed_practices: fresh = get_roadmap_item(tenant_id=tenant_id, item_id=gate_id) or updated practice_result = activate_stage_practice_set( tenant_id=tenant_id, initiative_id=initiative_id, stage=fresh, user_id=user_id, ) return { "gate_id": gate_id, "gate_title": updated.get("title"), "practice_transition": practice_result, } def _mark_join_reached( cur, *, tenant_id: str, join_id: str, ) -> None: cur.execute( """ UPDATE roadmap_items SET status = 'reached', updated_at = NOW() WHERE id = %s AND tenant_id = %s AND item_type = 'join_gate' """, (join_id, tenant_id), ) def try_switch_join( *, tenant_id: str, initiative_id: str, join_id: str, user_id: Optional[str] = None, ) -> Optional[dict[str, Any]]: from services.audit import log_audit from services.roadmap import get_roadmap_item, list_dependencies_for_initiative join = get_roadmap_item(tenant_id=tenant_id, item_id=join_id) if not join or not is_join_gate(join): return None if join.get("status") in TERMINAL_STATUSES: return None dependencies = list_dependencies_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) item_by_id = { str(item["id"]): item for item in __import__( "services.roadmap", fromlist=["list_roadmap_items_for_initiative"] ).list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) } pending = [ input_id for input_id in join_input_gate_ids(join_id, dependencies) if item_by_id.get(input_id, {}).get("status") != "reached" ] if pending: return None output_ids = join_output_gate_ids(join_id, dependencies) activated: list[dict[str, Any]] = [] conn = get_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: _mark_join_reached(cur, tenant_id=tenant_id, join_id=join_id) conn.commit() finally: conn.close() join_id_str = str(join_id) if join_id_str in item_by_id: item_by_id[join_id_str] = {**item_by_id[join_id_str], "status": "reached"} for output_id in output_ids: output_gate = item_by_id.get(output_id) if not output_gate or not is_work_gate(output_gate): continue if output_gate.get("status") in TERMINAL_STATUSES: continue if not _graph_ready(output_id, item_by_id=item_by_id, dependencies=dependencies): continue activated.append( activate_work_gate( tenant_id=tenant_id, initiative_id=initiative_id, gate=output_gate, user_id=user_id, ) ) log_audit( "progression.join_switched", user_id=user_id, tenant_id=tenant_id, details={ "initiative_id": initiative_id, "join_id": join_id, "join_title": join.get("title"), "activated_gate_ids": [item["gate_id"] for item in activated], }, ) return { "join_id": join_id, "join_title": join.get("title"), "activated": activated, } def try_switch_joins_for_gate( *, tenant_id: str, initiative_id: str, gate_id: str, user_id: Optional[str] = None, ) -> list[dict[str, Any]]: from services.roadmap import list_dependencies_for_initiative, list_roadmap_items_for_initiative dependencies = list_dependencies_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) joins = [ item for item in list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) if is_join_gate(item) ] switched: list[dict[str, Any]] = [] for join in joins: join_id = str(join["id"]) inputs = join_input_gate_ids(join_id, dependencies) if gate_id not in inputs: continue result = try_switch_join( tenant_id=tenant_id, initiative_id=initiative_id, join_id=join_id, user_id=user_id, ) if result: switched.append(result) return switched def _legacy_linear_next_stage( *, tenant_id: str, initiative_id: str, reached_item_id: str, user_id: Optional[str] = None, ) -> Optional[dict[str, Any]]: from services.roadmap import get_roadmap_item, list_roadmap_items_for_initiative, update_roadmap_item stages = [ item for item in list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) if is_work_gate(item) ] stages.sort(key=lambda item: (item.get("sort_order", 0), item.get("title", ""))) reached_index = next( (index for index, stage in enumerate(stages) if stage["id"] == reached_item_id), None, ) if reached_index is None: return None next_stage = None for stage in stages[reached_index + 1 :]: if stage.get("status") in {"planned", "active", "at_risk"}: next_stage = stage break if not next_stage: return None update_roadmap_item( tenant_id=tenant_id, item_id=next_stage["id"], status="active", user_id=user_id, ) fresh = get_roadmap_item(tenant_id=tenant_id, item_id=next_stage["id"]) if not fresh: return None return activate_work_gate( tenant_id=tenant_id, initiative_id=initiative_id, gate=fresh, user_id=user_id, ) def on_work_gate_reached( *, tenant_id: str, initiative_id: str, reached_item_id: str, user_id: Optional[str] = None, ) -> dict[str, Any]: from services.maturity_practice import list_practices_for_stage, update_recurring_element from services.roadmap import get_roadmap_item, list_dependencies_for_initiative, list_roadmap_items_for_initiative reached = get_roadmap_item(tenant_id=tenant_id, item_id=reached_item_id) if not reached or not is_work_gate(reached): return {"reached_gate_id": reached_item_id} for practice in list_practices_for_stage( tenant_id=tenant_id, initiative_id=initiative_id, stage_id=reached_item_id, ): if practice.get("status") == "active": update_recurring_element( tenant_id=tenant_id, recurring_id=practice["id"], status="paused", user_id=user_id, ) dependencies = list_dependencies_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) has_progression_deps = any( (dep.get("dependency_type") or "requires") == "requires" for dep in dependencies if str(dep["from_item_id"]) in { str(reached_item_id), *(str(item["id"]) for item in list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) if is_join_gate(item)), } or str(dep["to_item_id"]) == str(reached_item_id) ) joined = try_switch_joins_for_gate( tenant_id=tenant_id, initiative_id=initiative_id, gate_id=reached_item_id, user_id=user_id, ) activated_successors: list[dict[str, Any]] = [] if not joined: item_by_id = { str(item["id"]): item for item in list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) } for successor_id in successor_work_gate_ids(reached_item_id, dependencies): successor = item_by_id.get(successor_id) if not successor or not is_work_gate(successor): continue if is_join_gate(successor): continue if successor.get("status") in TERMINAL_STATUSES: continue if not _graph_ready( successor_id, item_by_id=item_by_id, dependencies=dependencies ): continue activated_successors.append( activate_work_gate( tenant_id=tenant_id, initiative_id=initiative_id, gate=successor, user_id=user_id, ) ) legacy = None if not joined and not activated_successors and not has_progression_deps: legacy = _legacy_linear_next_stage( tenant_id=tenant_id, initiative_id=initiative_id, reached_item_id=reached_item_id, user_id=user_id, ) return { "reached_gate_id": reached_item_id, "joins_switched": joined, "successors_activated": activated_successors, "legacy_linear": legacy, }