"""HCM:21 phase management per quarter. Each quarter cycles through four phases: Scanning → Planning → Producing → Controlling Each phase has minimum action requirements before the next phase unlocks. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, List, Set PHASES = ["scanning", "planning", "producing", "controlling"] PHASE_ACTIONS: Dict[str, List[str]] = { "scanning": [ "query_department", "query_employees", "calculate_metric", "review_financials", ], "planning": [ "set_hiring_target", "set_training_budget", "set_compensation_policy", "set_retention_program", ], "producing": [ "execute_hiring", "execute_promotion", "execute_transfer", "execute_training", "execute_termination", ], "controlling": [ "submit_report", "calculate_metric", ], } # Minimum actions required in a phase before advancing PHASE_MIN_ACTIONS = { "scanning": 2, "planning": 1, "producing": 1, "controlling": 1, } # Control actions available in any phase ALWAYS_AVAILABLE = ["advance_phase", "advance_quarter"] @dataclass class PhaseManager: """Tracks phase state within a quarter.""" current_phase: str = "scanning" actions_in_phase: int = 0 phase_history: Dict[str, int] = field(default_factory=lambda: {p: 0 for p in PHASES}) phases_completed: Set[str] = field(default_factory=set) def get_available_actions(self) -> List[str]: """Get list of valid action types for the current phase.""" actions = list(PHASE_ACTIONS.get(self.current_phase, [])) # advance_phase available if minimum actions met if self.can_advance_phase(): actions.append("advance_phase") # advance_quarter available if we're in controlling phase with min actions met if self.current_phase == "controlling" and self.actions_in_phase >= PHASE_MIN_ACTIONS["controlling"]: actions.append("advance_quarter") # submit_report only in controlling return actions def is_valid_action(self, action_type: str) -> bool: """Check if an action type is valid in the current phase.""" return action_type in self.get_available_actions() def can_advance_phase(self) -> bool: """Check if the minimum action requirement is met for advancing.""" min_required = PHASE_MIN_ACTIONS.get(self.current_phase, 1) return self.actions_in_phase >= min_required def advance_phase(self) -> str: """Move to the next phase. Returns new phase name. Raises ValueError if minimum actions not met or already in last phase. """ if not self.can_advance_phase(): min_req = PHASE_MIN_ACTIONS.get(self.current_phase, 1) raise ValueError( f"Cannot advance: need {min_req} actions in {self.current_phase} phase, " f"only {self.actions_in_phase} taken." ) current_idx = PHASES.index(self.current_phase) if current_idx >= len(PHASES) - 1: raise ValueError("Already in the final phase (controlling). Use advance_quarter instead.") self.phases_completed.add(self.current_phase) self.current_phase = PHASES[current_idx + 1] self.actions_in_phase = 0 return self.current_phase def record_action(self, action_type: str) -> None: """Record that an action was taken in the current phase.""" if action_type not in ("advance_phase", "advance_quarter"): self.actions_in_phase += 1 self.phase_history[self.current_phase] = self.phase_history.get(self.current_phase, 0) + 1 def reset_for_new_quarter(self) -> None: """Reset phase state for a new quarter.""" self.current_phase = "scanning" self.actions_in_phase = 0 self.phase_history = {p: 0 for p in PHASES} self.phases_completed = set() def get_phase_summary(self) -> dict: """Get summary of phase progress.""" return { "current_phase": self.current_phase, "actions_in_phase": self.actions_in_phase, "min_required": PHASE_MIN_ACTIONS.get(self.current_phase, 1), "can_advance": self.can_advance_phase(), "phases_completed": sorted(self.phases_completed), "phase_history": dict(self.phase_history), }